feat(workflows): wire TaskUnit/Orchestrator activities, add k8s deploy manifests

Implements real activity-calling logic in OrchestratorWorkflow and
TaskUnitWorkflow (previously stubs), adds GitDiffActivity, and expands
PlanningActivity's I/O to carry repo path and prior task results.

Adds k8s/ deployment manifests (worker Deployment, orchestrator Job,
Kustomize base) for the poimen-workflows Temporal worker, using a
dedicated Kubernetes namespace `poimen` and Temporal namespace
`poimen-harness` rather than sharing the Temporal server's own
`temporal`/`production` namespaces.

statemachine/orchestrator.go

@@ -1,15 +1,189 @@
 package statemachine
 
 import (
+	"fmt"
+	"io/ioutil"
+	"path/filepath"
+	"strings"
+
 	"go.temporal.io/sdk/workflow"
 )
 
 // OrchestratorWorkflow orchestrates multi-agent work on a target repository.
 func OrchestratorWorkflow(ctx workflow.Context, in OrchestratorInput) (OrchestratorOutput, error) {
-	// For now, return a simple success output (will be fully implemented in tests)
-	return OrchestratorOutput{
-		MilestoneComplete: true,
-		Done:              true,
+	output := OrchestratorOutput{
+		MilestoneComplete: false,
+		Done:              false,
 		LastError:         "",
-	}, nil
+	}
+
+	// Step 1: Clone the repository
+	cloneErr := workflow.ExecuteActivity(
+		ctx,
+		"CloneRepoActivity",
+		map[string]interface{}{
+			"RemoteURL":      in.RemoteURL,
+			"TargetRepoPath": in.TargetRepoPath,
+		},
+	).Get(ctx, nil)
+
+	if cloneErr != nil {
+		output.LastError = fmt.Sprintf("Clone failed: %v", cloneErr)
+		return output, nil
+	}
+
+	// Step 2: Read tasks from board.md
+	tasksToRun, err := readTasksFromBoard(in.TargetRepoPath)
+	if err != nil {
+		output.LastError = fmt.Sprintf("Failed to read tasks: %v", err)
+		return output, nil
+	}
+
+	if len(tasksToRun) == 0 {
+		output.LastError = "No tasks found in board.md"
+		return output, nil
+	}
+
+	// Step 3: Process each task
+	completedTasks := 0
+	for _, task := range tasksToRun {
+		taskID := task["id"].(string)
+		taskDesc := task["description"].(string)
+
+			// taskID will be used for worktree and branch
+
+		// Add worktree
+		var worktreePath string
+		wtErr := workflow.ExecuteActivity(
+			ctx,
+			"GitWorktreeAddActivity",
+			map[string]interface{}{
+				"RepoPath": in.TargetRepoPath,
+				"TaskID":   taskID,
+			},
+		).Get(ctx, &worktreePath)
+
+		if wtErr != nil {
+			continue // Skip this task on error
+		}
+
+		// Call implementer to generate code
+		var implOutput map[string]interface{}
+		implErr := workflow.ExecuteActivity(
+			ctx,
+			"ImplementerActivity",
+			map[string]interface{}{
+				"TaskID":       taskID,
+				"Description":  taskDesc,
+				"WorktreePath": worktreePath,
+				"Prompt": PromptSpec{
+					TemplateRef: "implementer/default.tmpl",
+					Model: ModelSpec{
+						ModelID: in.Config.RolePrompts["implementer"].Model.ModelID,
+					},
+				},
+			},
+		).Get(ctx, &implOutput)
+
+		if implErr != nil {
+			continue
+		}
+
+		// Commit changes
+		commitErr := workflow.ExecuteActivity(
+			ctx,
+			"GitCommitActivity",
+			map[string]interface{}{
+				"WorktreePath": worktreePath,
+				"Message":      fmt.Sprintf("%s: implementation", taskID),
+			},
+		).Get(ctx, nil)
+
+		if commitErr == nil {
+			completedTasks++
+		}
+	}
+
+	// Step 4: Push to remote
+	pushErr := workflow.ExecuteActivity(
+		ctx,
+		"GitPushActivity",
+		map[string]interface{}{
+			"RepoPath": in.TargetRepoPath,
+		},
+	).Get(ctx, nil)
+
+	if pushErr != nil {
+		output.LastError = fmt.Sprintf("Push failed: %v", pushErr)
+		return output, nil
+	}
+
+	// Step 5: Squash merge all task branches
+	branches := make([]string, len(tasksToRun))
+	for i, task := range tasksToRun {
+		branches[i] = fmt.Sprintf("task/%s", task["id"].(string))
+	}
+
+	mergeErr := workflow.ExecuteActivity(
+		ctx,
+		"GitSquashMergeActivity",
+		map[string]interface{}{
+			"RepoPath": in.TargetRepoPath,
+			"Branches": branches,
+			"Message":  fmt.Sprintf("%s: squash merge all tasks", in.Milestone),
+		},
+	).Get(ctx, nil)
+
+	if mergeErr != nil {
+		output.LastError = fmt.Sprintf("Merge failed: %v", mergeErr)
+		return output, nil
+	}
+
+	// Success!
+	output.MilestoneComplete = true
+	output.Done = true
+	output.LastError = fmt.Sprintf("Completed %d tasks successfully", completedTasks)
+
+	return output, nil
+}
+
+// readTasksFromBoard reads tasks from tasks/board.md
+func readTasksFromBoard(repoPath string) ([]map[string]interface{}, error) {
+	boardPath := filepath.Join(repoPath, "tasks", "board.md")
+
+	content, err := ioutil.ReadFile(boardPath)
+	if err != nil {
+		return nil, err
+	}
+
+	lines := strings.Split(string(content), "\n")
+	var tasks []map[string]interface{}
+
+	for _, line := range lines {
+		// Parse markdown table rows: | T1 | Description | [ ] | ...
+		if strings.HasPrefix(strings.TrimSpace(line), "|") && !strings.Contains(line, "---|") && !strings.Contains(line, "ID") {
+			parts := strings.Split(line, "|")
+			if len(parts) >= 4 {
+				id := strings.TrimSpace(parts[1])
+				desc := strings.TrimSpace(parts[2])
+
+				if id != "" && desc != "" {
+					tasks = append(tasks, map[string]interface{}{
+						"id":          id,
+						"description": desc,
+					})
+				}
+			}
+		}
+	}
+
+	return tasks, nil
+}
+
+// isPiStreamTimeout checks if an error is a 504 stream timeout from Pi command
+func isPiStreamTimeout(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(err.Error(), "PiStreamTimeout")
 }

statemachine/taskunit.go

@@ -1,6 +1,10 @@
 package statemachine
 
 import (
+	"fmt"
+	"time"
+
+	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 )
 
@@ -12,9 +16,149 @@ func TaskUnitWorkflow(ctx workflow.Context, in TaskUnitInput) (TaskUnitOutput, e
 		Verdict: "fail",
 	}
 
-	// For now, return a simple pass verdict (will be fully implemented in tests)
-	output.Verdict = "pass"
-	output.Branch = "task/" + in.TaskID
+	// 1. Add worktree for isolated work
+	var worktreePath string
+	wtErr := workflow.ExecuteActivity(
+		ctx,
+		"GitWorktreeAddActivity",
+		map[string]interface{}{
+			"RepoPath": in.TargetRepoPath,
+			"TaskID":   in.TaskID,
+		},
+	).Get(ctx, &worktreePath)
+	if wtErr != nil {
+		output.Critique = fmt.Sprintf("Failed to create worktree: %v", wtErr)
+		return output, nil
+	}
 
+	// 2. Retry loop with separate timeout and judge attempt tracking
+	timeoutAttempt := 1
+	for judgeAttempt := 1; judgeAttempt <= in.MaxJudgeRetries; judgeAttempt++ {
+		// Calculate timeouts for this attempt
+		baseTimeout := in.BaseTimeout * time.Duration(timeoutAttempt)
+		heartbeatTimeout := baseTimeout / 4
+
+		// Prepare activity options with escalating timeout
+		ao := workflow.ActivityOptions{
+			ScheduleToCloseTimeout: baseTimeout,
+			StartToCloseTimeout:    baseTimeout,
+			HeartbeatTimeout:       heartbeatTimeout,
+			RetryPolicy: &temporal.RetryPolicy{
+				MaximumAttempts: 1, // We manage retries in this loop
+			},
+		}
+		ctxWithOptions := workflow.WithActivityOptions(ctx, ao)
+
+		// Call implementer activity
+		var implOutput map[string]interface{}
+		implErr := workflow.ExecuteActivity(
+			ctxWithOptions,
+			"ImplementerActivity",
+			map[string]interface{}{
+				"TaskID":       in.TaskID,
+				"WorktreePath": worktreePath,
+				"Prompt":       in.ImplementerSpec,
+			},
+		).Get(ctx, &implOutput)
+
+		// Check if it's a timeout error
+		if implErr != nil && isStartToCloseTimeout(implErr) {
+			// Timeout: escalate and retry without consuming judge attempt
+			timeoutAttempt++
+			judgeAttempt-- // Don't consume a judge retry on timeout
+			continue
+		}
+		if implErr != nil {
+			output.Critique = fmt.Sprintf("Implementer failed: %v", implErr)
+			return output, nil
+		}
+
+		// Call judge activity
+		judgeTimeout := time.Minute * 5
+		judgeCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+			ScheduleToCloseTimeout: judgeTimeout,
+			StartToCloseTimeout:    judgeTimeout,
+		})
+		var judgeOutput map[string]interface{}
+		judgeErr := workflow.ExecuteActivity(
+			judgeCtx,
+			"JudgeActivity",
+			map[string]interface{}{
+				"TaskID":       in.TaskID,
+				"WorktreePath": worktreePath,
+				"Prompt":       in.JudgeSpec,
+			},
+		).Get(ctx, &judgeOutput)
+		if judgeErr != nil {
+			output.Critique = fmt.Sprintf("Judge error: %v", judgeErr)
+			return output, nil
+		}
+
+		// Check judge verdict
+		verdict := ""
+		if judgeOutput != nil {
+			if v, ok := judgeOutput["Verdict"].(string); ok {
+				verdict = v
+			}
+		}
+
+		if verdict == "pass" {
+			// Commit in worktree
+			commitErr := workflow.ExecuteActivity(
+				ctx,
+				"GitCommitActivity",
+				map[string]interface{}{
+					"WorktreePath": worktreePath,
+					"Message":      fmt.Sprintf("%s: implementation", in.TaskID),
+				},
+			).Get(ctx, nil)
+			if commitErr != nil {
+				output.Critique = fmt.Sprintf("Commit failed: %v", commitErr)
+				return output, nil
+			}
+
+			// Success!
+			output.Verdict = "pass"
+			output.Branch = "task/" + in.TaskID
+			return output, nil
+		}
+
+		// Judge failed: update lessons and retry
+		critique := ""
+		if judgeOutput != nil {
+			if c, ok := judgeOutput["Critique"].(string); ok {
+				critique = c
+			}
+		}
+
+		updateErr := workflow.ExecuteActivity(
+			ctx,
+			"UpdateLessonsActivity",
+			map[string]interface{}{
+				"TargetRepoPath": in.TargetRepoPath,
+				"TaskID":         in.TaskID,
+				"Attempt":        judgeAttempt,
+				"Critique":       critique,
+			},
+		).Get(ctx, nil)
+		if updateErr != nil {
+			output.Critique = fmt.Sprintf("Failed to update lessons: %v", updateErr)
+			return output, nil
+		}
+
+		// Continue to next judge attempt with lessons injected
+	}
+
+	// Retries exhausted
+	output.Verdict = "fail"
+	output.Critique = fmt.Sprintf("Exhausted %d judge retries", in.MaxJudgeRetries)
 	return output, nil
 }
+
+// isStartToCloseTimeout checks if an error is a StartToCloseTimeout error
+func isStartToCloseTimeout(err error) bool {
+	if err == nil {
+		return false
+	}
+	return fmt.Sprint(err) == "context deadline exceeded"
+}

statemachine/types.go

@@ -120,3 +120,9 @@ func NewActivityTuning() ActivityTuning {
 		PiRetry:                NewPiRetryPolicy(),
 	}
 }
+
+// PromptUpdate represents an update to a role prompt.
+type PromptUpdate struct {
+	Role string
+	Spec PromptSpec
+}