diff --git a/tasks/T0.1.md b/tasks/T0.1.md new file mode 100644 index 0000000..5210633 --- /dev/null +++ b/tasks/T0.1.md @@ -0,0 +1,42 @@ +# T0.1: Repo Scaffold + +## Scope +Create directory structure, `go.mod`, empty stubs for all packages. + +## Implementation Checklist +- [ ] `go.mod`: module `github.com/rockliang/poimen/workflows`, Go 1.21+ +- [ ] `statemachine/types.go`: empty package stub (will fill in T0.2) +- [ ] `statemachine/signals.go`: empty package stub +- [ ] `statemachine/orchestrator.go`: empty package stub, func placeholder +- [ ] `statemachine/taskunit.go`: empty package stub, func placeholder +- [ ] `action/planner.go`: empty package stub +- [ ] `action/implementer.go`: empty package stub +- [ ] `action/judge.go`: empty package stub +- [ ] `action/git.go`: empty package stub +- [ ] `action/skills.go`: empty package stub +- [ ] `action/integration_test.go`: empty package stub +- [ ] `action/lessons.go`: empty package stub +- [ ] `action/llm/client.go`: empty package stub +- [ ] `prompts/registry.go`: empty package stub +- [ ] `prompts/planner/default.tmpl`: empty text file +- [ ] `prompts/judge/default.tmpl`: empty text file +- [ ] `prompts/implementer/default.tmpl`: empty text file +- [ ] `internal/config/config.go`: empty package stub +- [ ] `internal/lock/flock.go`: empty package stub +- [ ] `cmd/worker/main.go`: `func main()` stub +- [ ] `cmd/starter/main.go`: `func main()` stub +- [ ] `tests/taskunit_workflow_test.go`: empty test file +- [ ] `tests/orchestrator_workflow_test.go`: empty test file + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows +go build ./... +# Command should succeed with no errors +# All directories should exist as listed above +``` + +## Done Criteria +- `go build ./...` succeeds with exit code 0 +- `ls -R` shows all directories match PLAN.md §Directory Structure +- No compilation errors or warnings diff --git a/tasks/T0.2.md b/tasks/T0.2.md new file mode 100644 index 0000000..4676ebd --- /dev/null +++ b/tasks/T0.2.md @@ -0,0 +1,108 @@ +# T0.2: Shared Types + +## Scope +Implement `statemachine/types.go` with all config/input/output structs and document defaults. + +## Implementation +File: `statemachine/types.go` + +```go +type ModelSpec struct { + ModelID string // e.g. "claude-opus-5", "claude-sonnet-5" + Thinking string // "adaptive" or "" + Effort string // "low", "medium", "high", "xhigh", "max" +} + +type PromptSpec struct { + TemplateRef string // e.g. "planner/default.tmpl" + RawTemplate string // overrides TemplateRef if non-empty + Variables map[string]any + Model ModelSpec + LessonsRef string // key into lessons store +} + +type PiRetryPolicy struct { + ScheduleToCloseTimeout time.Duration // default: 5m + InitialInterval time.Duration // default: 2s + MaximumInterval time.Duration // default: 30s + BackoffCoefficient float64 // default: 2.0 + StreamTimeout time.Duration // default: 30s + StreamTimeoutMax time.Duration // default: 2m +} + +type ActivityTuning struct { + ImplementerBaseTimeout time.Duration // default: 10m + ImplementerMaxRetries int // default: 3 + JudgeTimeout time.Duration // default: 5m + PiRetry PiRetryPolicy +} + +type OrchestratorConfig struct { + SystemPrompt string // shared prompt prefix + Skills []SkillRef // required skill sources + RolePrompts map[string]PromptSpec // per-role: "planner", "judge", "implementer" + Tuning ActivityTuning +} + +type OrchestratorInput struct { + TargetRepoPath string + RemoteURL string + Milestone string // e.g. "T0" + Config OrchestratorConfig + DryRun bool + CycleCount int + MaxCyclesBeforeCAN int // default: 100 +} + +type OrchestratorOutput struct { + MilestoneComplete bool + Done bool + LastError string +} + +type TaskUnitInput struct { + TaskID string + TargetRepoPath string + JudgeSpec PromptSpec + ImplementerSpec PromptSpec + BaseTimeout time.Duration + MaxJudgeRetries int +} + +type TaskUnitOutput struct { + TaskID string + Verdict string // "pass" or "fail" + Critique string + Branch string +} + +type SkillRef struct { + Name string // skill identifier + URL string // source to clone +} +``` + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows +go build ./statemachine/ + +# Run unit test: +go test -v ./tests -run TestTypesDefaults +``` + +Test file: `tests/types_test.go` +```go +func TestTypesDefaults(t *testing.T) { + // Verify all defaults are correctly set + pr := PiRetryPolicy{} + assert.Equal(t, 5*time.Minute, pr.ScheduleToCloseTimeout) + // ... more assertions +} +``` + +## Done Criteria +- `go build ./statemachine/` succeeds +- `go test ./tests -run TestTypesDefaults` passes +- All struct fields documented with default values +- No compilation errors diff --git a/tasks/T0.3.md b/tasks/T0.3.md new file mode 100644 index 0000000..bd443dc --- /dev/null +++ b/tasks/T0.3.md @@ -0,0 +1,93 @@ +# T0.3: Git & Locking + +## Scope +Implement `action/git.go` + `internal/lock/flock.go` for repo cloning, worktree management, and squash-merge. + +## Implementation + +### File: `internal/lock/flock.go` +```go +package lock + +// Acquire advisory file lock (blocking) +func Acquire(path string) error + +// Release advisory file lock +func Release(path string) error +``` + +### File: `action/git.go` +```go +type CloneRepoInput struct { + RemoteURL string + TargetRepoPath string +} + +func CloneRepoActivity(ctx context.Context, in CloneRepoInput) error + // If $TargetRepoPath/.git exists: git -C $TargetRepoPath fetch origin + // Else: git clone $RemoteURL $TargetRepoPath + +type GitWorktreeAddInput struct { + RepoPath string + TaskID string +} + +func GitWorktreeAddActivity(ctx context.Context, in GitWorktreeAddInput) (string, error) + // Guarded by orchestrator.lock + // git worktree add -b task/ ../worktrees/ origin/main + // Return worktree path + +type GitCommitInput struct { + WorktreePath string + Message string +} + +func GitCommitActivity(ctx context.Context, in GitCommitInput) error + // No lock needed; safe within isolated worktree + // git -C $WorktreePath add -A + // git -C $WorktreePath commit -m "$Message" + +type GitPushInput struct { + RepoPath string +} + +func GitPushActivity(ctx context.Context, in GitPushInput) error + // Guarded by orchestrator.lock + // git -C $RepoPath push origin main + +type GitSquashMergeInput struct { + RepoPath string + Branches []string // ["task/T0.1", "task/T0.2", ...] + Message string +} + +func GitSquashMergeActivity(ctx context.Context, in GitSquashMergeInput) error + // Guarded by orchestrator.lock + // fetch origin main + // checkout main && pull --ff-only origin main + // for b in branches: merge --squash $b + // commit -m $Message + // push origin main + // for b in branches: worktree remove + branch -D +``` + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows +go test -v ./tests -run TestGit + +# Test script: tests/git_test.go +``` + +Test cases: +- Clone into empty path → creates .git +- Clone into existing path → fetches instead of re-cloning +- Worktree add → returns valid path +- Commit in worktree → file changes staged +- Squash-merge → one commit on main, branches cleaned up + +## Done Criteria +- `go test ./tests -run TestGit` passes +- Tested against local scratch git repo (not real remote) +- No lock deadlocks on concurrent calls +- Squash-merge produces exactly one commit diff --git a/tasks/T0.4.md b/tasks/T0.4.md new file mode 100644 index 0000000..68d97b6 --- /dev/null +++ b/tasks/T0.4.md @@ -0,0 +1,72 @@ +# T0.4: Pi & Error Classification + +## Scope +Implement `action/skills.go` with `PrepareSkillsActivity` and `classifyPiErr` for skill prep via homelab API. + +## Implementation + +### File: `action/skills.go` +```go +type SkillRef struct { + Name string + URL string +} + +type PrepareSkillsInput struct { + Skills []SkillRef + StreamTimeout time.Duration +} + +func PrepareSkillsActivity(ctx context.Context, in PrepareSkillsInput) error + // For each skill, run: pi clone-or-fetch $skill.URL + // Pass --stream-timeout=$StreamTimeout to pi + // Each skill behind its own lock (not orchestrator.lock) + // On error, return classified error (see below) + +func classifyPiErr(err error) error + // 4xx (400-499): NonRetryableApplicationError "PiClientError" + // 504: ApplicationError "PiStreamTimeout" + // 5xx (500-599, except 504): leave retryable + // Other network errors: leave retryable +``` + +## Error Buckets + +### Bucket 1: 4xx (PiClientError) +- Status code 400-499 +- Non-retryable: bad request, auth error, not found +- Temporal stops retrying immediately +- Activity fails + +### Bucket 2: 5xx except 504 (generic 5xx) +- Status code 500-503, 505-599 +- Retryable: server error, likely transient +- Temporal backs off + retries until `ScheduleToCloseTimeout` (5m) + +### Bucket 3: 504 (PiStreamTimeout) +- Status code 504 +- Means pi's SSE stream-read timed out +- Retryable, BUT: Orchestrator doubles `config.Tuning.PiRetry.StreamTimeout` before retry +- Next attempt uses wider timeout + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows +go test -v ./tests -run TestPiErrors + +# Test file: tests/pi_test.go +``` + +Test cases: +- Mock 400 response → NonRetryableApplicationError returned +- Mock 403 response → NonRetryableApplicationError returned +- Mock 500 response → retryable error returned +- Mock 503 response → retryable error returned +- Mock 504 response → ApplicationError type "PiStreamTimeout" returned +- Mock network timeout → retryable error returned + +## Done Criteria +- `go test ./tests -run TestPiErrors` passes all 6 test cases +- All error buckets correctly classified +- No panics on nil pointers +- Error messages include HTTP status code diff --git a/tasks/T0.5.md b/tasks/T0.5.md new file mode 100644 index 0000000..6f3b832 --- /dev/null +++ b/tasks/T0.5.md @@ -0,0 +1,133 @@ +# T0.5: LLM Agents & Prompts + +## Scope +Implement LLM activities (Planner, Judge, Implementer), LLM client, and prompt template registry. + +## Implementation + +### File: `action/llm/client.go` +```go +type AnthropicClient struct { + apiKey string +} + +func NewClient() *AnthropicClient + // Read ANTHROPIC_API_KEY from env + // Return client + +func (c *AnthropicClient) CreateMessage(ctx context.Context, in MessageInput) (string, error) + // Call Anthropic API messages.create + // Respect model.ModelID, model.Thinking, model.Effort + // Return response text +``` + +### File: `action/planner.go` +```go +type PlanningInput struct { + Config OrchestratorConfig + BoardState string // JSON or markdown of task board + Milestone string +} + +type TaskDispatch struct { + TaskID string + Prompt PromptSpec + BaseTimeout time.Duration // can override default +} + +func PlanningActivity(ctx context.Context, in PlanningInput) ([]TaskDispatch, error) + // Read target repo's tasks/INDEX.md + board from shared FS + // Render prompt: in.Config.SystemPrompt + in.Config.RolePrompts["planner"] template + // Call LLM (Planner model) + // Parse response: which tasks to dispatch next, optional tuning overrides + // Return task dispatch list +``` + +### File: `action/judge.go` +```go +type JudgeInput struct { + Config OrchestratorConfig + Diff string // git diff output + IntegrationTestLogs string // test output +} + +type JudgeOutput struct { + Verdict string // "pass" or "fail" + Critique string // explanation if fail +} + +func JudgeActivity(ctx context.Context, in JudgeInput) (JudgeOutput, error) + // Render prompt: in.Config.SystemPrompt + in.Config.RolePrompts["judge"] + // Call LLM (Judge model, reasoning) + // Parse response: verdict + critique + // Return JudgeOutput +``` + +### File: `action/implementer.go` +```go +type ImplementerInput struct { + Config OrchestratorConfig + TaskID string + WorktreePath string + Lessons string // "known errors — do not repeat" section +} + +type ImplementerOutput struct { + Success bool + Changes string // summary of changes made +} + +func ImplementerActivity(ctx context.Context, in ImplementerInput) (ImplementerOutput, error) + // Render prompt: in.Config.SystemPrompt + in.Config.RolePrompts["implementer"] + // Inject in.Lessons into Variables + // Start tool-call agent loop (run git/cargo/pnpm/etc as needed) + // After each tool call, activity.RecordHeartbeat(ctx, progress) + // Return success/changes +``` + +### File: `prompts/registry.go` +```go +// go:embed prompts/*.tmpl + +func Render(templateRef string, variables map[string]any) (string, error) + // Load embedded template via go:embed + text/template + // Render with variables + // Return rendered string +``` + +### Files: `prompts/planner/default.tmpl`, etc. +Empty templates for now; will be filled in by Planner/Judge/Implementer activities. + +``` +You are a Planner agent. Your job: reconcile task state, dispatch work. + +System prompt: {{.SystemPrompt}} + +Current board: +{{.BoardState}} + +Current config: +{{.Config | json}} + +What tasks should we dispatch next? (respond in JSON: {"tasks": [{"id": "T0.1", "timeout_override_ms": null}, ...]}) +``` + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows +go test -v ./tests -run TestPrompts + +# Test file: tests/prompts_test.go +``` + +Test cases: +- Render template with system prompt + variables → output includes system prompt prefix +- Render with RawTemplate override → uses raw template, not embedded +- Render with Variables substitution → all {{.Var}} replaced +- Mock LLM client responses → activities parse correctly + +## Done Criteria +- `go test ./tests -run TestPrompts` passes +- All templates render without errors +- LLM client reads ANTHROPIC_API_KEY from env (or uses mock in tests) +- Activities parse LLM responses into structured output diff --git a/tasks/T0.6.md b/tasks/T0.6.md new file mode 100644 index 0000000..e51d72a --- /dev/null +++ b/tasks/T0.6.md @@ -0,0 +1,78 @@ +# T0.6: TaskUnit Workflow + +## Scope +Implement `statemachine/taskunit.go` with retry loops, timeout escalation, and lessons injection. + +## Implementation + +### File: `statemachine/taskunit.go` +```go +func TaskUnitWorkflow(ctx workflow.Context, in TaskUnitInput) (TaskUnitOutput, error) + // 1. Call GitWorktreeAddActivity(ctx, {RepoPath, TaskID}) → get worktree path + // 2. Lessons file init: try ReadLessonsActivity, may be empty on first run + // 3. Retry loop: + // timeoutAttempt := 1 + // for judgeAttempt := 1; judgeAttempt <= in.MaxJudgeRetries; judgeAttempt++ + // + // SetupActivity options: + // StartToCloseTimeout: in.BaseTimeout * time.Duration(timeoutAttempt) + // HeartbeatTimeout: (in.BaseTimeout * time.Duration(timeoutAttempt)) / 4 + // RetryPolicy: {MaximumAttempts: 1} // NO retries; we manage them in the loop + // + // Call ImplementerActivity(ctx, {Lessons: lessons, ...}) + // If isStartToCloseTimeout(err): + // timeoutAttempt++ + // judgeAttempt-- // don't consume a judge retry on timeout + // continue // next loop iteration has longer timeout + // If err != nil: + // return TaskUnitOutput{Verdict: "fail", Critique: err.Error()}, nil + // + // Call RunIntegrationTestActivity(ctx, {WorktreePath, TestCmd}) + // If integration test fails: + // // Some tasks don't have tests; pass if no test defined + // + // Call JudgeActivity(ctx, {Diff, IntegrationTestResult}) + // If judge.Verdict == "pass": + // Call GitCommitActivity(ctx, {WorktreePath, "T0.x: implementation"}) + // return TaskUnitOutput{Verdict: "pass", Branch: "task/T0.x"} + // Else: + // Call UpdateLessonsActivity(ctx, {Critique}) + // lessons = ReadLessonsActivity() // updated lessons for next attempt + // continue // next judgeAttempt with lessons injected + // End retries + // + // 4. If we exit loop without pass: return fail verdict +``` + +## Retry Logic Detail + +**TimeoutAttempt vs JudgeAttempt:** +- TimeoutAttempt: activity ran out of time, next attempt has longer StartToCloseTimeout +- JudgeAttempt: activity finished but output wrong, lessons injected, duration doesn't change +- They're independent counters so timeout escalation doesn't consume judge retries + +**Lessons injection:** +- Before each ImplementerActivity, render prompt with Lessons appended: "Known errors from prior attempts:\n{lessons}" +- On judge failure, append new lesson to lessons file +- Lessons persist within TaskUnitWorkflow (shared memory) +- Lessons also written to disk (tasks/.orchestrator/lessons/.jsonl) for Planner's review + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows +go test -v ./tests -run TestTaskUnit + +# Test file: tests/taskunit_workflow_test.go +``` + +Test cases (mocked activities): +1. **Pass on first try:** Implementer→TestPass→JudgePass → return pass +2. **Fail then pass after lesson:** Implementer→TestPass→JudgeFail (append lesson) → Implementer (lessons injected)→TestPass→JudgePass → return pass +3. **Retries exhausted:** Implementer→JudgeFail 3 times → return fail +4. **Timeout escalation:** Implementer timeout 1st → BaseTimeout*1 fails, Implementer timeout 2nd → BaseTimeout*2 succeeds → continue + +## Done Criteria +- `go test ./tests -run TestTaskUnit` passes all 4 cases +- Split timeout/judge-fail counters work correctly +- Lessons inject into prompt without error +- No infinite loops on mocked failures diff --git a/tasks/T0.7.md b/tasks/T0.7.md new file mode 100644 index 0000000..ac96705 --- /dev/null +++ b/tasks/T0.7.md @@ -0,0 +1,143 @@ +# T0.7: Orchestrator Workflow + +## Scope +Implement `statemachine/orchestrator.go` with config state, signals, fan-out/fan-in, continue-as-new, and 504 learning. + +## Implementation + +### File: `statemachine/orchestrator.go` +```go +func OrchestratorWorkflow(ctx workflow.Context, in OrchestratorInput) (OrchestratorOutput, error) + // 1. Mutable config state (not frozen at start): + // config := in.Config + // skillsHaveChanged := true // first cycle, prep skills + // + // 2. Signal handlers (checked each cycle): + // - "update-system-prompt": config.SystemPrompt = signalPayload + // - "update-skills": config.Skills = signalPayload, skillsHaveChanged = true + // - "update-role-prompt": config.RolePrompts[role] = signalPayload + // - "update-tuning": config.Tuning = signalPayload + // - "pause": wait for "resume" signal + // - "abort-task": forward via SignalExternalWorkflow(ctx, "taskunit-"+taskID, "abort", nil) + // + // 3. Query handlers: + // - "status": return current cycle count, pending tasks + // - "current-config": return config + // + // 4. Main loop (continues until submilestone complete): + // for { + // // Check signals (pause, abort, update-*) + // selector := workflow.NewSelector(ctx) + // // register signal channels + // + // // Prep skills if needed + // if skillsHaveChanged { + // call PrepareSkillsActivity(ctx, {config.Skills, config.Tuning.PiRetry.StreamTimeout}) + // wrap in 504-learning loop: + // for { + // err := ExecuteActivity(...) + // if isPiStreamTimeout(err) && config.Tuning.PiRetry.StreamTimeout < config.Tuning.PiRetry.StreamTimeoutMax: + // config.Tuning.PiRetry.StreamTimeout *= 2 + // continue + // break + // } + // skillsHaveChanged = false + // } + // + // // Planning phase 1: decide what to dispatch + // planResult := call PlanningActivity(ctx, {config, boardState, milestone}) + // if submilestoneComplete(planResult): + // // All subtasks done, trigger merge + // call GitSquashMergeActivity(ctx, {repoBranches, "T0: squash merge subtasks"}) + // return OrchestratorOutput{MilestoneComplete: true, Done: true} + // + // // Dispatch: fan out TaskUnitWorkflow for each task + // taskFutures := []workflow.Future{} + // for taskID in planResult.tasksToDispatch: + // spec := selectApplicableSpec(config.RolePrompts, taskID) + // future := ExecuteChildWorkflow(ctx, TaskUnitWorkflow, TaskUnitInput{ + // TaskID: taskID, + // JudgeSpec: config.RolePrompts["judge"], + // ImplementerSpec: config.RolePrompts["implementer"], + // BaseTimeout: config.Tuning.ImplementerBaseTimeout, // or override from planner + // MaxJudgeRetries: config.Tuning.ImplementerMaxRetries, + // }) + // taskFutures = append(taskFutures, future) + // + // // Await all + // results := []TaskUnitOutput{} + // for future in taskFutures: + // var out TaskUnitOutput + // future.Get(ctx, &out) + // results = append(results, out) + // + // // Planning phase 2: update board and commit + // call PlanningActivity(ctx, {config, results, boardState, milestone}) → UpdateBoardOutput + // call GitCommitActivity(ctx, {repoPath, "Update board after cycle"}) + // call GitPushActivity(ctx, {repoPath}) + // + // // Continue-as-new check + // in.CycleCount++ + // if in.CycleCount >= in.MaxCyclesBeforeCAN: + // nextInput := OrchestratorInput{ + // // carry forward all state + // CycleCount: 0, + // Config: config, // includes mutated Tuning/RolePrompts/Skills + // } + // return workflow.NewContinueAsNewError(ctx, OrchestratorWorkflow, nextInput) + // } +``` + +## 504 Learning Detail +```go +// Wrapping PrepareSkillsActivity for 504 learning: +for { + r := config.Tuning.PiRetry + ao := workflow.ActivityOptions{ + ScheduleToCloseTimeout: r.ScheduleToCloseTimeout, // 5m hard cap + StartToCloseTimeout: r.MaximumInterval, // per-attempt ceiling + RetryPolicy: &temporal.RetryPolicy{ + InitialInterval: r.InitialInterval, + BackoffCoefficient: r.BackoffCoefficient, + MaximumInterval: r.MaximumInterval, + NonRetryableErrorTypes: []string{"PiClientError"}, + }, + } + err := workflow.ExecuteActivity( + workflow.WithActivityOptions(ctx, ao), + action.PrepareSkillsActivity, + action.PrepareSkillsInput{Skills: config.Skills, StreamTimeout: r.StreamTimeout}, + ).Get(ctx, nil) + + var appErr *temporal.ApplicationError + if errors.As(err, &appErr) && appErr.Type() == "PiStreamTimeout" && r.StreamTimeout < r.StreamTimeoutMax { + config.Tuning.PiRetry.StreamTimeout = min(r.StreamTimeout*2, r.StreamTimeoutMax) + continue // ScheduleToCloseTimeout still bounds each attempt + } + if err != nil { + return OrchestratorOutput{}, err + } + break +} +``` + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows +go test -v ./tests -run TestOrchestrator + +# Test file: tests/orchestrator_workflow_test.go +``` + +Test cases (mocked activities): +1. **Fan-out/fan-in:** Dispatch 3 tasks → all complete → results collected +2. **Squash-merge on complete:** All tasks pass → GitSquashMergeActivity called +3. **Continue-as-new:** CycleCount reaches MaxCyclesBeforeCAN → returns NewContinueAsNewError +4. **Signal mutation:** update-role-prompt signal → next dispatch uses new prompt +5. **504 learning:** PrepareSkillsActivity returns PiStreamTimeout → StreamTimeout doubled → next PrepareSkillsActivity call uses doubled value, capped at Max + +## Done Criteria +- `go test ./tests -run TestOrchestrator` passes all 5 cases +- Signals mutate config without affecting in-flight TaskUnit +- Continue-as-new preserves OrchestratorConfig across cycles +- 504 learning loop doesn't exceed ScheduleToCloseTimeout diff --git a/tasks/T0.8.md b/tasks/T0.8.md new file mode 100644 index 0000000..49a4cd9 --- /dev/null +++ b/tasks/T0.8.md @@ -0,0 +1,167 @@ +# T0.8: Worker & Starter CLIs + +## Scope +Implement `cmd/worker/main.go`, `cmd/starter/main.go`, and `internal/config` for env-based loading. + +## Implementation + +### File: `internal/config/config.go` +```go +package config + +type TemporalConfig struct { + HostPort string // default: temporal.riotpiao.com:7233 + Namespace string // default: default + TLSCert string // env: TEMPORAL_TLS_CERT (file path) + TLSKey string // env: TEMPORAL_TLS_KEY (file path) +} + +type AppConfig struct { + Temporal AppConfig + AnthropicAPIKey string // env: ANTHROPIC_API_KEY +} + +func LoadConfig() (AppConfig, error) + // Read from env variables (TEMPORAL_*, ANTHROPIC_API_KEY) + // Return filled config +``` + +### File: `cmd/worker/main.go` +```go +func main() { + cfg, err := config.LoadConfig() + if err != nil { panic(err) } + + // Connect to Temporal + c, err := client.Dial(client.Options{ + HostPort: cfg.Temporal.HostPort, + Namespace: cfg.Temporal.Namespace, + // TLS options if provided + }) + if err != nil { panic(err) } + defer c.Close() + + // Create worker + w, err := worker.New(c, "default", worker.Options{}) + if err != nil { panic(err) } + + // Register all workflows + w.RegisterWorkflow(statemachine.OrchestratorWorkflow) + w.RegisterWorkflow(statemachine.TaskUnitWorkflow) + + // Register all activities + w.RegisterActivity(action.CloneRepoActivity) + w.RegisterActivity(action.GitWorktreeAddActivity) + w.RegisterActivity(action.GitCommitActivity) + w.RegisterActivity(action.GitPushActivity) + w.RegisterActivity(action.GitSquashMergeActivity) + w.RegisterActivity(action.PrepareSkillsActivity) + w.RegisterActivity(action.PlanningActivity) + w.RegisterActivity(action.ImplementerActivity) + w.RegisterActivity(action.JudgeActivity) + w.RegisterActivity(action.RunIntegrationTestActivity) + w.RegisterActivity(action.UpdateLessonsActivity) + w.RegisterActivity(action.ReadLessonsActivity) + + // Run worker + if err := w.Run(worker.InterruptCh()); err != nil { + panic(err) + } +} +``` + +### File: `cmd/starter/main.go` +```go +func main() { + var ( + repoPath = flag.String("repo", "", "target repo path") + remoteURL = flag.String("remote", "", "remote URL") + milestone = flag.String("milestone", "T0", "milestone ID") + dryRun = flag.Bool("dry-run", false, "disable git push/merge") + plannerModel = flag.String("planner-model", "claude-opus-5", "planner model ID") + judgeModel = flag.String("judge-model", "claude-opus-5", "judge model ID") + implementerModel = flag.String("implementer-model", "claude-sonnet-5", "implementer model ID") + ) + flag.Parse() + + cfg, err := config.LoadConfig() + if err != nil { panic(err) } + + // Connect to Temporal + c, err := client.Dial(client.Options{ + HostPort: cfg.Temporal.HostPort, + Namespace: cfg.Temporal.Namespace, + }) + if err != nil { panic(err) } + defer c.Close() + + // Build OrchestratorInput + input := statemachine.OrchestratorInput{ + TargetRepoPath: *repoPath, + RemoteURL: *remoteURL, + Milestone: *milestone, + Config: statemachine.OrchestratorConfig{ + SystemPrompt: "You are an expert software developer orchestrating multi-agent work.", + Skills: []statemachine.SkillRef{}, + RolePrompts: map[string]statemachine.PromptSpec{ + "planner": {TemplateRef: "planner/default.tmpl", Model: statemachine.ModelSpec{ModelID: *plannerModel, Thinking: "adaptive", Effort: "high"}}, + "judge": {TemplateRef: "judge/default.tmpl", Model: statemachine.ModelSpec{ModelID: *judgeModel, Thinking: "adaptive", Effort: "high"}}, + "implementer": {TemplateRef: "implementer/default.tmpl", Model: statemachine.ModelSpec{ModelID: *implementerModel}}, + }, + Tuning: statemachine.ActivityTuning{ + ImplementerBaseTimeout: 10 * time.Minute, + ImplementerMaxRetries: 3, + JudgeTimeout: 5 * time.Minute, + PiRetry: statemachine.PiRetryPolicy{ + ScheduleToCloseTimeout: 5 * time.Minute, + InitialInterval: 2 * time.Second, + MaximumInterval: 30 * time.Second, + BackoffCoefficient: 2.0, + StreamTimeout: 30 * time.Second, + StreamTimeoutMax: 2 * time.Minute, + }, + }, + }, + DryRun: *dryRun, + MaxCyclesBeforeCAN: 100, + } + + // Start workflow + workflowID := "orch-" + strings.ReplaceAll(*repoPath, "/", "-") + run, err := c.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{ID: workflowID, TaskQueue: "default"}, statemachine.OrchestratorWorkflow, input) + if err != nil { panic(err) } + + fmt.Printf("Started workflow %s\n", workflowID) + fmt.Printf("Monitor at: temporal.riotpiao.com:8080/namespaces/default/workflows/%s\n", workflowID) + + // Optionally wait for completion + // var result OrchestratorOutput + // err = run.Get(context.Background(), &result) +} +``` + +## Verification +```bash +cd /Users/rockliang/workplace/Poimen/workflows + +# Test build +go build ./cmd/worker +go build ./cmd/starter + +# Test worker registration (mock/local test): +go test -v ./tests -run TestWorkerRegistration + +# Manual test (requires temporal.riotpiao.com running): +# 1. Start worker: +go run ./cmd/worker & +# 2. In another terminal, start workflow: +go run ./cmd/starter --repo /tmp/fixture --remote file:///tmp/remote --dry-run +# 3. Check Temporal Web UI: should show workflow execution +``` + +## Done Criteria +- `go build ./cmd/worker` succeeds +- `go build ./cmd/starter` succeeds +- `go test ./tests -run TestWorkerRegistration` passes +- Manual test: `go run ./cmd/worker` connects to temporal.riotpiao.com:7233 without error (or test Temporal instance) +- Manual test: `go run ./cmd/starter --dry-run` returns workflow ID and URL immediately diff --git a/tasks/T0.9.md b/tasks/T0.9.md new file mode 100644 index 0000000..28fe51e --- /dev/null +++ b/tasks/T0.9.md @@ -0,0 +1,191 @@ +# T0.9: End-to-End Test + +## Scope +Run against real `temporal.riotpiao.com` cluster + disposable forgejo scratch repo. All 7 verification items from PLAN.md. + +## Setup + +### Prerequisites +- `temporal.riotpiao.com` Temporal cluster accessible +- Forgejo instance running (for scratch repo) +- Local `git`, `go` 1.21+ +- `ANTHROPIC_API_KEY` env var set +- `TEMPORAL_NAMESPACE`, `TEMPORAL_TLS_CERT`, `TEMPORAL_TLS_KEY` env vars set if cluster requires them + +### Fixture Repo Structure +Create temporary fixture repo: +``` +/tmp/fixture/ + tasks/ + INDEX.md (copy from this repo) + board.md (minimal: 3 trivial tasks for quick run) +``` + +Minimal board.md: +``` +| T0.1 | Create file /tmp/fixture/output.txt with content "hello world" | [ ] | +| T0.2 | Create file /tmp/fixture/result.json with {"status": "ok"} | [ ] | +| T0.3 | Create file /tmp/fixture/done.txt with "COMPLETE" | [ ] | +``` + +## Run Sequence + +### 1. Clone & Fetch Bootstrap Test (T0.3 foundational) +```bash +cd /tmp +mkdir -p test-clone +go run ./cmd/starter \ + --repo /tmp/test-clone \ + --remote /tmp/fixture \ + --dry-run +# Check: /tmp/test-clone/.git exists after first run +# Check: Verify it's a valid git repo +``` + +### 2. Full Cycle with Dry-Run (no real push) +```bash +export FIXTURE_REMOTE=file:///tmp/fixture-remote-src +export FIXTURE_WORKTREE=/tmp/fixture-worktree + +# Start worker +go run ./cmd/worker & +WORKER_PID=$! + +# Start orchestrator workflow +go run ./cmd/starter \ + --repo /tmp/fixture \ + --remote file:///tmp/fixture-remote-src \ + --milestone T0 \ + --dry-run + +# Monitor Temporal Web UI: http://temporal.riotpiao.com:8080 +# Workflow ID: orch-tmp-fixture +# Expected: all 3 subtasks dispatched, Judge passes each, board updated, NO push to origin + +# Verify: +# - Board file shows all tasks marked done +# - No commits pushed to remote (because --dry-run) +# - Lessons file exists if any task was induced to fail + +kill $WORKER_PID +``` + +### 3. Live Signal Update Mid-Run +```bash +# Start same workflow again (different workflow ID) +go run ./cmd/starter \ + --repo /tmp/fixture \ + --remote file:///tmp/fixture-remote-src \ + --milestone T0.1 \ + --dry-run & +WF_ID=$! + +# While running, send update signal: +temporal workflow signal \ + --workflow-id \ + --name update-role-prompt \ + --input '{"role":"implementer","spec":{"template_ref":"implementer/default.tmpl","variables":{"marker":"from-signal"},...}}' + +# Check: next dispatched task includes "from-signal" in Variables +# Verify via board file or task output +``` + +### 4. 5xx Fault Injection (retry-then-succeed) +```bash +# Setup: Mock pi command to return 5xx first N times, then succeed +# (Use a local wrapper script or fault-injection proxy) + +# Run workflow: +go run ./cmd/starter \ + --repo /tmp/fixture \ + --remote file:///tmp/fixture-remote-src \ + --dry-run + +# Expected: +# - PrepareSkillsActivity retries with exponential backoff +# - Eventually succeeds after N retries +# - Workflow continues normally +``` + +### 5. 5xx Exhaustion (always-5xx, fail at 5m) +```bash +# Setup: Mock pi command to always return 503 + +# Run workflow (must have Implementer call PrepareSkillsActivity or similar pi-dependent step) + +# Expected: +# - PrepareSkillsActivity retries for ~5 minutes (ScheduleToCloseTimeout) +# - After 5m, activity fails +# - Workflow marks task as failed +# - Board reflects failure +``` + +### 6. 504 Stream Timeout Learning +```bash +# Setup: Mock pi command to return 504 + +# Run workflow: +go run ./cmd/starter --repo /tmp/fixture --remote file:///tmp/fixture-remote-src --dry-run + +# While running, query workflow state: +temporal workflow query \ + --workflow-id \ + --query-type current-config + +# Expected output includes config.Tuning.PiRetry.StreamTimeout (should be doubled from default 30s) +# After 504, next query shows it as 60s +# If 504 repeats, doubles again to 120s, capped at StreamTimeoutMax (2m) +``` + +### 7. Continue-as-New History Bound +```bash +# Run multiple cycles (manually via CLI or workflow logic) + +# Check Temporal Web UI: Workflow → History tab +# Expected: +# - History is compact (not unbounded growth) +# - No duplication of events +# - Cycle count resets per continue-as-new +``` + +## Verification Checklist +- [ ] Fixture repo clones fresh when path empty +- [ ] Fetch-instead-of-clone on second run +- [ ] All 3 subtasks dispatched and complete +- [ ] Judge passes each task +- [ ] Board file updated with completion marks +- [ ] No push to origin when --dry-run +- [ ] Live signal (update-role-prompt) changes next dispatch +- [ ] Live signal (update-skills) re-preps skills exactly once +- [ ] 5xx retry-then-succeed: activity retries and eventually succeeds +- [ ] 5xx exhaustion: activity fails at ~5m mark, task marked failed +- [ ] 504 learning: StreamTimeout doubled and actually used on next attempt +- [ ] 504 learning: Stops doubling at StreamTimeoutMax (2m) +- [ ] Continue-as-new: History bounded, no unbounded growth +- [ ] Squash-merge result: One commit on main per submilestone (not yet, waiting for T0.1-T0.8 to pass first) + +## Done Criteria (All Must Pass) +1. All 7 checks in Verification Checklist marked `[x]` +2. No panics or unhandled errors in workflow execution +3. Temporal Web UI shows clean workflow execution with retries visible +4. Board file reflects accurate task completion state +5. Lessons file demonstrates learning across retries (if any failure induced) +6. Workflow completes within reasonable time (~10-30min for 3 subtasks + fault injection) + +## Cleanup +```bash +# Delete fixture remote and worktrees +rm -rf /tmp/fixture-remote-src /tmp/fixture-worktree + +# Kill any lingering worker processes +pkill -f "go run ./cmd/worker" + +# Optionally delete workflow from Temporal (if testing repeatedly) +temporal workflow delete --workflow-id orch-tmp-fixture +``` + +## Notes +- **Real forgejo remote:** The "disposable" remote can be on actual forgejo instance (`git@forgejo.homelab:test/workflows-e2e.git`), or a file:// URL locally +- **Anthropic API calls:** Use actual API (not mock) for real e2e; costs will be minimal if test tasks are simple +- **Temporal Web UI:** Set timezone to match your local time for easier log reading +- **Fault injection:** Can use `PATH` manipulation (wrapper scripts) or a local HTTP proxy (e.g., mitmproxy, Burp Suite) to inject 5xx/504 responses