kurihada/ai-workflow-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add orch strict worktree dispatch

Browse Source
This commit is contained in:
kurihada
2026-03-19 13:52:51 +08:00
parent 07f4a6fdae
commit 1b0cd723d7
9 changed files with 776 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cli/orch/dispatch.go
+8 -9
View File
@@ -16,6 +16,7 @@ type dispatchOptions struct {
body string
bodyFile string
baseRef string
repoPath string
workspaceRoot string
strictWorktree bool
}
@@ -27,10 +28,6 @@ func newDispatchCmd(root *rootOptions) *cobra.Command {
Use: "dispatch",
Short: "Dispatch a ready task to a worker through inbox",
RunE: func(cmd *cobra.Command, args []string) error {
if opts.workspaceRoot != "" || opts.strictWorktree {
return protocol.InvalidInput("worktree dispatch is not implemented yet", nil)
}
body, err := resolveBodyValue(opts.body, opts.bodyFile)
if err != nil {
return err
@@ -44,11 +41,12 @@ func newDispatchCmd(root *rootOptions) *cobra.Command {
defer sqlDB.Close()
result, err := store.NewOrchStore(sqlDB).DispatchTask(ctx, store.DispatchInput{
RunID: opts.runID,
TaskID: opts.taskID,
ToAgent: opts.toAgent,
Body: body,
BaseRef: opts.baseRef,
RunID: opts.runID,
TaskID: opts.taskID,
ToAgent: opts.toAgent,
Body: body,
BaseRef: opts.baseRef,
PrepareWorkspace: newDispatchWorkspacePreparer(cmd, *opts),
})
if err != nil {
return err
@@ -85,6 +83,7 @@ func newDispatchCmd(root *rootOptions) *cobra.Command {
cmd.Flags().StringVar(&opts.body, "body", "", "Task message body")
cmd.Flags().StringVar(&opts.bodyFile, "body-file", "", "Read task message body from file")
cmd.Flags().StringVar(&opts.baseRef, "base-ref", "", "Optional base ref to record on the attempt")
cmd.Flags().StringVar(&opts.repoPath, "repo-path", "", "Source repository path for worktree dispatch")
cmd.Flags().StringVar(&opts.workspaceRoot, "workspace-root", "", "Workspace root for worktree dispatch")
cmd.Flags().BoolVar(&opts.strictWorktree, "strict-worktree", false, "Require strict worktree setup")
_ = cmd.MarkFlagRequired("run")
internal/cli/orch/git_test_helpers_test.go
+53
View File
@@ -0,0 +1,53 @@
package orch
import (
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
)
func initGitRepo(t *testing.T) string {
t.Helper()
repoPath := filepath.Join(t.TempDir(), "repo")
if err := os.MkdirAll(repoPath, 0o755); err != nil {
t.Fatalf("mkdir repo path: %v", err)
}
runGitCommand(t, repoPath, "init")
runGitCommand(t, repoPath, "config", "user.email", "test@example.com")
runGitCommand(t, repoPath, "config", "user.name", "Test User")
readmePath := filepath.Join(repoPath, "README.md")
if err := os.WriteFile(readmePath, []byte("hello\n"), 0o644); err != nil {
t.Fatalf("write README.md: %v", err)
}
runGitCommand(t, repoPath, "add", "README.md")
runGitCommand(t, repoPath, "commit", "-m", "init")
return repoPath
}
func gitHeadCommit(t *testing.T, repoPath string) string {
t.Helper()
cmd := exec.Command("git", "-C", repoPath, "rev-parse", "--verify", "HEAD^{commit}")
output, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("git rev-parse HEAD in %s: %v\n%s", repoPath, err, output)
}
return strings.TrimSpace(string(output))
}
func runGitCommand(t *testing.T, repoPath string, args ...string) {
t.Helper()
cmd := exec.Command("git", append([]string{"-C", repoPath}, args...)...)
output, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("git %v in %s: %v\n%s", args, repoPath, err, output)
}
}
internal/cli/orch/integration_test.go
+222
View File
@@ -1,6 +1,7 @@
package orch
import (
"os"
"path/filepath"
"testing"
)
@@ -504,3 +505,224 @@ func TestOrchDispatchRejectsNonReadyTask(t *testing.T) {
}
assertErrorJSON(t, stdout, "invalid_state")
}
func TestOrchDispatchCreatesStrictWorktree(t *testing.T) {
t.Parallel()
dbPath := filepath.Join(t.TempDir(), "coord.db")
repoPath := initGitRepo(t)
runOrchCommand(
t,
"--db", dbPath,
"--json",
"run", "init",
"--run", "run_blog_worktree_001",
"--goal", "Validate strict worktree dispatch",
)
runOrchCommand(
t,
"--db", dbPath,
"--json",
"task", "add",
"--run", "run_blog_worktree_001",
"--task", "T1",
"--title", "Implement backend",
"--default-to", "worker-a",
)
dispatchOut := runOrchCommand(
t,
"--db", dbPath,
"--json",
"dispatch",
"--run", "run_blog_worktree_001",
"--task", "T1",
"--repo-path", repoPath,
"--workspace-root", ".orch/worktrees",
"--strict-worktree",
"--body", "Implement inside isolated worktree.",
)
var dispatchResp map[string]any
mustDecodeJSON(t, dispatchOut, &dispatchResp)
attempt, ok := nestedValue(t, dispatchResp, "data", "attempt").(map[string]any)
if !ok {
t.Fatalf("expected attempt object, got %#v", nestedValue(t, dispatchResp, "data", "attempt"))
}
if got, _ := attempt["base_ref"].(string); got != "HEAD" {
t.Fatalf("expected base_ref HEAD, got %#v", attempt["base_ref"])
}
expectedCommit := gitHeadCommit(t, repoPath)
if got, _ := attempt["base_commit"].(string); got != expectedCommit {
t.Fatalf("expected base_commit %q, got %#v", expectedCommit, attempt["base_commit"])
}
if got, _ := attempt["branch_name"].(string); got != "orch/run-blog-worktree-001/T1/attempt-1" {
t.Fatalf("unexpected branch name %#v", attempt["branch_name"])
}
worktreePath, _ := attempt["worktree_path"].(string)
if worktreePath == "" {
t.Fatalf("expected worktree_path, got %#v", attempt["worktree_path"])
}
if got, _ := attempt["workspace_status"].(string); got != "created" {
t.Fatalf("expected workspace_status created, got %#v", attempt["workspace_status"])
}
if _, err := os.Stat(worktreePath); err != nil {
t.Fatalf("stat worktree path %s: %v", worktreePath, err)
}
if _, err := os.Stat(filepath.Join(worktreePath, "README.md")); err != nil {
t.Fatalf("expected README.md in worktree: %v", err)
}
message, ok := nestedValue(t, dispatchResp, "data", "message").(map[string]any)
if !ok {
t.Fatalf("expected message object, got %#v", nestedValue(t, dispatchResp, "data", "message"))
}
payload, ok := message["payload_json"].(map[string]any)
if !ok {
t.Fatalf("expected payload_json object, got %#v", message["payload_json"])
}
if got, _ := payload["worktree_path"].(string); got != worktreePath {
t.Fatalf("expected payload worktree path %q, got %#v", worktreePath, payload["worktree_path"])
}
threadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id")
runInboxCommand(
t,
"--db", dbPath,
"--json",
"claim",
"--agent", "worker-a",
"--thread", threadID,
)
runInboxCommand(
t,
"--db", dbPath,
"--json",
"update",
"--agent", "worker-a",
"--thread", threadID,
"--status", "in_progress",
"--summary", "Started inside worktree",
)
reconcileOut := runOrchCommand(
t,
"--db", dbPath,
"--json",
"reconcile",
"--run", "run_blog_worktree_001",
)
var reconcileResp map[string]any
mustDecodeJSON(t, reconcileOut, &reconcileResp)
updatedTasks := nestedArray(t, reconcileResp, "data", "updated_tasks")
if len(updatedTasks) != 1 {
t.Fatalf("expected one updated task after worktree reconcile, got %#v", updatedTasks)
}
}
func TestOrchStrictWorktreeRejectsDirtyRepoWithoutBaseRef(t *testing.T) {
t.Parallel()
dbPath := filepath.Join(t.TempDir(), "coord.db")
repoPath := initGitRepo(t)
if err := os.WriteFile(filepath.Join(repoPath, "dirty.txt"), []byte("dirty\n"), 0o644); err != nil {
t.Fatalf("write dirty file: %v", err)
}
runOrchCommand(
t,
"--db", dbPath,
"--json",
"run", "init",
"--run", "run_blog_worktree_002",
"--goal", "Validate dirty repo rejection",
)
runOrchCommand(
t,
"--db", dbPath,
"--json",
"task", "add",
"--run", "run_blog_worktree_002",
"--task", "T1",
"--title", "Implement backend",
"--default-to", "worker-a",
)
stdout, _, exitCode := executeOrchCommand(
"--db", dbPath,
"--json",
"dispatch",
"--run", "run_blog_worktree_002",
"--task", "T1",
"--repo-path", repoPath,
"--workspace-root", ".orch/worktrees",
"--strict-worktree",
)
if exitCode != 30 {
t.Fatalf("expected invalid_state exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
}
assertErrorJSON(t, stdout, "invalid_state")
if _, err := os.Stat(filepath.Join(repoPath, ".orch", "worktrees", "run_blog_worktree_002", "T1", "attempt-1")); !os.IsNotExist(err) {
t.Fatalf("expected no worktree directory on strict failure, got err=%v", err)
}
}
func TestOrchStrictWorktreeAllowsExplicitBaseRefOnDirtyRepo(t *testing.T) {
t.Parallel()
dbPath := filepath.Join(t.TempDir(), "coord.db")
repoPath := initGitRepo(t)
baseCommit := gitHeadCommit(t, repoPath)
if err := os.WriteFile(filepath.Join(repoPath, "dirty.txt"), []byte("dirty\n"), 0o644); err != nil {
t.Fatalf("write dirty file: %v", err)
}
runOrchCommand(
t,
"--db", dbPath,
"--json",
"run", "init",
"--run", "run_blog_worktree_003",
"--goal", "Validate explicit base ref on dirty repo",
)
runOrchCommand(
t,
"--db", dbPath,
"--json",
"task", "add",
"--run", "run_blog_worktree_003",
"--task", "T1",
"--title", "Implement backend",
"--default-to", "worker-a",
)
dispatchOut := runOrchCommand(
t,
"--db", dbPath,
"--json",
"dispatch",
"--run", "run_blog_worktree_003",
"--task", "T1",
"--repo-path", repoPath,
"--workspace-root", ".orch/worktrees",
"--strict-worktree",
"--base-ref", "HEAD",
)
var dispatchResp map[string]any
mustDecodeJSON(t, dispatchOut, &dispatchResp)
if got := nestedString(t, dispatchResp, "data", "attempt", "base_ref"); got != "HEAD" {
t.Fatalf("expected explicit base_ref HEAD, got %q", got)
}
if got := nestedString(t, dispatchResp, "data", "attempt", "base_commit"); got != baseCommit {
t.Fatalf("expected base_commit %q, got %q", baseCommit, got)
}
}
internal/cli/orch/worktree.go
+303
View File
@@ -0,0 +1,303 @@
package orch
import (
"context"
"fmt"
"os"
"os/exec"
"path/filepath"
"strings"
"ai-workflow-skill/internal/protocol"
"ai-workflow-skill/internal/store"
"github.com/spf13/cobra"
)
func newDispatchWorkspacePreparer(cmd *cobra.Command, opts dispatchOptions) store.DispatchWorkspacePreparer {
if !dispatchUsesWorktree(opts) {
return nil
}
ctx := cmd.Context()
return func(task store.Task, attemptNo int) (store.DispatchWorkspace, func(), error) {
return provisionDispatchWorkspace(ctx, opts, task, attemptNo)
}
}
func dispatchUsesWorktree(opts dispatchOptions) bool {
return strings.TrimSpace(opts.repoPath) != "" ||
strings.TrimSpace(opts.workspaceRoot) != "" ||
opts.strictWorktree
}
func provisionDispatchWorkspace(ctx context.Context, opts dispatchOptions, task store.Task, attemptNo int) (store.DispatchWorkspace, func(), error) {
repoRoot, err := resolveRepoRoot(ctx, opts.repoPath)
if err != nil {
return store.DispatchWorkspace{}, nil, err
}
workspaceRoot := resolveWorkspaceRoot(repoRoot, opts.workspaceRoot)
if err := ensureWorkspaceRootIgnored(repoRoot, workspaceRoot); err != nil {
return store.DispatchWorkspace{}, nil, err
}
baseRef, baseCommit, err := resolveDispatchBase(ctx, repoRoot, workspaceRoot, opts.baseRef, opts.strictWorktree)
if err != nil {
return store.DispatchWorkspace{}, nil, err
}
branchName := buildAttemptBranchName(task.RunID, task.TaskID, attemptNo)
worktreePath := buildAttemptWorktreePath(workspaceRoot, task.RunID, task.TaskID, attemptNo)
if err := os.MkdirAll(filepath.Dir(worktreePath), 0o755); err != nil {
return store.DispatchWorkspace{}, nil, fmt.Errorf("create worktree parent dir: %w", err)
}
if _, err := os.Stat(worktreePath); err == nil {
return store.DispatchWorkspace{}, nil, fmt.Errorf("%w: worktree path already exists: %s", store.ErrInvalidState, worktreePath)
} else if err != nil && !os.IsNotExist(err) {
return store.DispatchWorkspace{}, nil, fmt.Errorf("stat worktree path: %w", err)
}
if _, _, err := runGit(ctx, repoRoot, "worktree", "add", "-b", branchName, worktreePath, baseCommit); err != nil {
return store.DispatchWorkspace{}, nil, err
}
cleanup := func() {
_, _, _ = runGit(context.Background(), repoRoot, "worktree", "remove", "--force", worktreePath)
_, _, _ = runGit(context.Background(), repoRoot, "branch", "-D", branchName)
_ = os.RemoveAll(worktreePath)
}
return store.DispatchWorkspace{
BaseRef: baseRef,
BaseCommit: baseCommit,
BranchName: branchName,
WorktreePath: worktreePath,
WorkspaceStatus: "created",
}, cleanup, nil
}
func resolveRepoRoot(ctx context.Context, repoPath string) (string, error) {
startPath := strings.TrimSpace(repoPath)
if startPath == "" {
var err error
startPath, err = os.Getwd()
if err != nil {
return "", fmt.Errorf("get current working directory: %w", err)
}
}
absPath, err := filepath.Abs(startPath)
if err != nil {
return "", fmt.Errorf("resolve repo path: %w", err)
}
stdout, _, err := runGit(ctx, absPath, "rev-parse", "--show-toplevel")
if err != nil {
return "", protocol.InvalidInput("repo-path must point to a Git worktree", err)
}
return strings.TrimSpace(stdout), nil
}
func resolveDispatchBase(ctx context.Context, repoRoot, workspaceRoot, requestedBaseRef string, strict bool) (string, string, error) {
baseRef := strings.TrimSpace(requestedBaseRef)
if baseRef != "" {
baseCommit, err := resolveCommit(ctx, repoRoot, baseRef)
if err != nil {
return "", "", protocol.InvalidInput("base-ref must resolve to a commit", err)
}
return baseRef, baseCommit, nil
}
if strict {
dirty, err := repoHasUncommittedChanges(ctx, repoRoot, workspaceRoot)
if err != nil {
return "", "", err
}
if dirty {
return "", "", fmt.Errorf("%w: repository has uncommitted changes; specify --base-ref or clean the repo", store.ErrInvalidState)
}
}
baseCommit, err := resolveCommit(ctx, repoRoot, "HEAD")
if err != nil {
return "", "", protocol.InvalidInput("failed to resolve HEAD commit", err)
}
return "HEAD", baseCommit, nil
}
func resolveCommit(ctx context.Context, repoRoot, ref string) (string, error) {
stdout, _, err := runGit(ctx, repoRoot, "rev-parse", "--verify", ref+"^{commit}")
if err != nil {
return "", err
}
return strings.TrimSpace(stdout), nil
}
func repoHasUncommittedChanges(ctx context.Context, repoRoot, workspaceRoot string) (bool, error) {
stdout, _, err := runGit(ctx, repoRoot, "status", "--porcelain")
if err != nil {
return false, fmt.Errorf("check repository status: %w", err)
}
for _, line := range strings.Split(strings.TrimSpace(stdout), "\n") {
line = strings.TrimSpace(line)
if line == "" {
continue
}
if len(line) >= 3 {
path := strings.TrimSpace(line[3:])
if shouldIgnoreStatusPath(repoRoot, workspaceRoot, path) {
continue
}
}
return true, nil
}
return false, nil
}
func resolveWorkspaceRoot(repoRoot, configuredRoot string) string {
root := strings.TrimSpace(configuredRoot)
if root == "" {
return filepath.Join(repoRoot, ".orch", "worktrees")
}
if filepath.IsAbs(root) {
return root
}
return filepath.Join(repoRoot, root)
}
func ensureWorkspaceRootIgnored(repoRoot, workspaceRoot string) error {
relative, err := filepath.Rel(repoRoot, workspaceRoot)
if err != nil {
return fmt.Errorf("resolve workspace root exclude path: %w", err)
}
if relative == "." || strings.HasPrefix(relative, "..") {
return nil
}
pattern := filepath.ToSlash(relative)
if !strings.HasSuffix(pattern, "/") {
pattern += "/"
}
excludePath := filepath.Join(repoRoot, ".git", "info", "exclude")
if err := os.MkdirAll(filepath.Dir(excludePath), 0o755); err != nil {
return fmt.Errorf("create git info dir: %w", err)
}
content, err := os.ReadFile(excludePath)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("read git exclude file: %w", err)
}
if strings.Contains(string(content), pattern) {
return nil
}
appendContent := pattern + "\n"
if len(content) > 0 && !strings.HasSuffix(string(content), "\n") {
appendContent = "\n" + appendContent
}
file, err := os.OpenFile(excludePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
return fmt.Errorf("open git exclude file: %w", err)
}
defer file.Close()
if _, err := file.WriteString(appendContent); err != nil {
return fmt.Errorf("append git exclude file: %w", err)
}
return nil
}
func shouldIgnoreStatusPath(repoRoot, workspaceRoot, statusPath string) bool {
relative, err := filepath.Rel(repoRoot, workspaceRoot)
if err != nil || relative == "." || strings.HasPrefix(relative, "..") {
return false
}
relative = filepath.ToSlash(relative)
statusPath = filepath.ToSlash(strings.Trim(statusPath, `"`))
return statusPath == relative || strings.HasPrefix(statusPath, relative+"/")
}
func buildAttemptBranchName(runID, taskID string, attemptNo int) string {
return fmt.Sprintf(
"orch/%s/%s/attempt-%d",
sanitizeGitSegment(runID),
sanitizeGitSegment(taskID),
attemptNo,
)
}
func buildAttemptWorktreePath(workspaceRoot, runID, taskID string, attemptNo int) string {
return filepath.Join(
workspaceRoot,
sanitizePathSegment(runID),
sanitizePathSegment(taskID),
fmt.Sprintf("attempt-%d", attemptNo),
)
}
func sanitizeGitSegment(value string) string {
return sanitizeSegment(value)
}
func sanitizePathSegment(value string) string {
return sanitizeSegment(value)
}
func sanitizeSegment(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "item"
}
var b strings.Builder
lastDash := false
for _, r := range value {
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
b.WriteRune(r)
lastDash = false
continue
}
if r == '-' || r == '_' || r == '.' {
if !lastDash {
b.WriteByte('-')
lastDash = true
}
continue
}
if !lastDash {
b.WriteByte('-')
lastDash = true
}
}
result := strings.Trim(b.String(), "-.")
if result == "" {
return "item"
}
if strings.HasSuffix(result, ".lock") {
result = strings.TrimSuffix(result, ".lock") + "-lock"
}
return result
}
func runGit(ctx context.Context, repoRoot string, args ...string) (string, string, error) {
cmdArgs := append([]string{"-C", repoRoot}, args...)
cmd := exec.CommandContext(ctx, "git", cmdArgs...)
output, err := cmd.CombinedOutput()
if err == nil {
return string(output), "", nil
}
message := strings.TrimSpace(string(output))
if message == "" {
message = err.Error()
}
return "", message, fmt.Errorf("git %s: %s", strings.Join(args, " "), message)
}
internal/store/orch.go
+95 -31
View File
@@ -98,11 +98,12 @@ type ListReadyInput struct {
}
type DispatchInput struct {
RunID string
TaskID string
ToAgent string
Body string
BaseRef string
RunID string
TaskID string
ToAgent string
Body string
BaseRef string
PrepareWorkspace DispatchWorkspacePreparer
}
type DispatchResult struct {
@@ -118,6 +119,16 @@ type ReconcileResult struct {
UpdatedTasks []Task `json:"updated_tasks"`
}
type DispatchWorkspace struct {
BaseRef string `json:"base_ref,omitempty"`
BaseCommit string `json:"base_commit,omitempty"`
BranchName string `json:"branch_name,omitempty"`
WorktreePath string `json:"worktree_path,omitempty"`
WorkspaceStatus string `json:"workspace_status,omitempty"`
}
type DispatchWorkspacePreparer func(task Task, attemptNo int) (DispatchWorkspace, func(), error)
type BlockedTask struct {
Task Task `json:"task"`
Attempt TaskAttempt `json:"attempt"`
@@ -473,9 +484,29 @@ func (s *OrchStore) DispatchTask(ctx context.Context, input DispatchInput) (Disp
}
attemptNo := task.LatestAttemptNo + 1
workspace := DispatchWorkspace{
BaseRef: strings.TrimSpace(input.BaseRef),
}
cleanupWorkspace := func() {}
workspaceCommitted := false
if input.PrepareWorkspace != nil {
workspace, cleanupWorkspace, err = input.PrepareWorkspace(task, attemptNo)
if err != nil {
return DispatchResult{}, err
}
if cleanupWorkspace == nil {
cleanupWorkspace = func() {}
}
defer func() {
if !workspaceCommitted {
cleanupWorkspace()
}
}()
}
threadID := newID("thr")
messageID := newID("msg")
payloadJSON := buildDispatchPayload(task, attemptNo, input.BaseRef)
payloadJSON := buildDispatchPayload(task, attemptNo, workspace)
thread := Thread{
ThreadID: threadID,
RunID: task.RunID,
@@ -541,15 +572,19 @@ func (s *OrchStore) DispatchTask(ctx context.Context, input DispatchInput) (Disp
}
attempt := TaskAttempt{
RunID: task.RunID,
TaskID: task.TaskID,
AttemptNo: attemptNo,
AssignedTo: assignedTo,
ThreadID: threadID,
BaseRef: strings.TrimSpace(input.BaseRef),
Status: "dispatched",
CreatedAt: now,
UpdatedAt: now,
RunID: task.RunID,
TaskID: task.TaskID,
AttemptNo: attemptNo,
AssignedTo: assignedTo,
ThreadID: threadID,
BaseRef: workspace.BaseRef,
BaseCommit: workspace.BaseCommit,
BranchName: workspace.BranchName,
WorktreePath: workspace.WorktreePath,
WorkspaceStatus: workspace.WorkspaceStatus,
Status: "dispatched",
CreatedAt: now,
UpdatedAt: now,
}
_, err = tx.ExecContext(
ctx,
@@ -564,10 +599,10 @@ func (s *OrchStore) DispatchTask(ctx context.Context, input DispatchInput) (Disp
attempt.AssignedTo,
attempt.ThreadID,
nullIfEmpty(attempt.BaseRef),
nil,
nil,
nil,
nil,
nullIfEmpty(attempt.BaseCommit),
nullIfEmpty(attempt.BranchName),
nullIfEmpty(attempt.WorktreePath),
nullIfEmpty(attempt.WorkspaceStatus),
nil,
attempt.Status,
formatTime(attempt.CreatedAt),
@@ -613,6 +648,7 @@ func (s *OrchStore) DispatchTask(ctx context.Context, input DispatchInput) (Disp
if err := tx.Commit(); err != nil {
return DispatchResult{}, fmt.Errorf("commit dispatch transaction: %w", err)
}
workspaceCommitted = true
task.Status = "dispatched"
task.LatestAttemptNo = attempt.AttemptNo
@@ -704,9 +740,10 @@ func (s *OrchStore) ReconcileRun(ctx context.Context, runID string) (ReconcileRe
_, err = tx.ExecContext(
ctx,
`UPDATE task_attempts
SET status = ?, updated_at = ?
SET status = ?, workspace_status = COALESCE(?, workspace_status), updated_at = ?
WHERE run_id = ? AND task_id = ? AND attempt_no = ?`,
nextStatus,
nullIfEmpty(reconcileWorkspaceStatus(threadStatus)),
formatTime(now),
runID,
taskID,
@@ -1116,14 +1153,14 @@ func scanTask(scanner threadScanner) (Task, error) {
func scanAttempt(scanner threadScanner) (TaskAttempt, error) {
var (
attempt TaskAttempt
baseRef sql.NullString
baseCommit sql.NullString
branchName sql.NullString
worktreePath sql.NullString
workspaceStatus sql.NullString
resultCommit sql.NullString
createdAt, updated string
attempt TaskAttempt
baseRef sql.NullString
baseCommit sql.NullString
branchName sql.NullString
worktreePath sql.NullString
workspaceStatus sql.NullString
resultCommit sql.NullString
createdAt, updated string
)
if err := scanner.Scan(
@@ -1580,7 +1617,7 @@ func validateAndNormalizeJSONDefault(fieldName, value, defaultValue string) (str
return compact.String(), nil
}
func buildDispatchPayload(task Task, attemptNo int, baseRef string) string {
func buildDispatchPayload(task Task, attemptNo int, workspace DispatchWorkspace) string {
payload := map[string]any{
"run_id": task.RunID,
"task_id": task.TaskID,
@@ -1596,8 +1633,20 @@ func buildDispatchPayload(task Task, attemptNo int, baseRef string) string {
payload["acceptance"] = acceptance
}
}
if strings.TrimSpace(baseRef) != "" {
payload["base_ref"] = strings.TrimSpace(baseRef)
if strings.TrimSpace(workspace.BaseRef) != "" {
payload["base_ref"] = strings.TrimSpace(workspace.BaseRef)
}
if strings.TrimSpace(workspace.BaseCommit) != "" {
payload["base_commit"] = strings.TrimSpace(workspace.BaseCommit)
}
if strings.TrimSpace(workspace.BranchName) != "" {
payload["branch_name"] = strings.TrimSpace(workspace.BranchName)
}
if strings.TrimSpace(workspace.WorktreePath) != "" {
payload["worktree_path"] = strings.TrimSpace(workspace.WorktreePath)
}
if strings.TrimSpace(workspace.WorkspaceStatus) != "" {
payload["workspace_status"] = strings.TrimSpace(workspace.WorkspaceStatus)
}
return marshalJSON(payload)
@@ -1634,6 +1683,21 @@ func summarizeAnswer(body string) string {
return line
}
func reconcileWorkspaceStatus(threadStatus string) string {
switch threadStatus {
case "pending":
return "created"
case "claimed", "in_progress", "blocked":
return "active"
case "done", "failed":
return "completed"
case "cancelled":
return "abandoned"
default:
return ""
}
}
func isUniqueConstraintError(err error) bool {
return strings.Contains(strings.ToLower(err.Error()), "unique constraint failed")
}