diff --git a/.go-harness/workflows/ux-feedback-check/main.go b/.go-harness/workflows/ux-feedback-check/main.go new file mode 100644 index 0000000..7a7d4e8 --- /dev/null +++ b/.go-harness/workflows/ux-feedback-check/main.go @@ -0,0 +1,14 @@ +package main + +import sdk "go-agent-harness/pkg/workflowsdk" + +func main() { + sdk.Main(func(ctx *sdk.Context) (any, error) { + _ = ctx.Phase("Workflow UX") + _ = ctx.Log("workflow ux path running") + _ = ctx.Feedback("finding", "workflow feedback reached host", map[string]any{ + "path": "api-and-tmux", + }) + return map[string]any{"ok": true, "args": ctx.Args}, nil + }) +} diff --git a/.go-harness/workflows/ux-feedback-check/workflow.json b/.go-harness/workflows/ux-feedback-check/workflow.json new file mode 100644 index 0000000..c927245 --- /dev/null +++ b/.go-harness/workflows/ux-feedback-check/workflow.json @@ -0,0 +1,9 @@ +{ + "name": "ux-feedback-check", + "description": "Workflow UX smoke that emits phase, log, finding feedback, and returns args.", + "version": 1, + "language": "go", + "entrypoint": "main.go", + "when_to_use": "Use to validate workflow discovery, run, feedback, SSE, and result handling.", + "timeout_seconds": 30 +} diff --git a/AGENTS.md b/AGENTS.md index fe76650..7c76d23 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,6 +24,10 @@ For a fresh agent worktree, run `scripts/init.sh ` before doing anyth ## 3) Non-Negotiables (Detailed Rules in Runbooks) +- Communication preference: respond concisely but educationally. Explain what + changed and why it matters. When you struggle with something and solve it, + document the symptom, cause, and fix in the appropriate durable log or plan + note so future agents can reuse the learning. - Strict TDD and no trivial/underspecified tests: `docs/runbooks/testing.md` - Tests must pass before commit: `docs/runbooks/testing.md` - Never allow failures in the accepted baseline. A failing test, failing package, failing regression command, or known red verification step is a blocker, not acceptable "pre-existing" state. diff --git a/CLAUDE.md b/CLAUDE.md index 6516f2b..f8a03ed 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -52,6 +52,9 @@ and error recovery chains. ## Operational Reminder +- Respond concisely but educationally: explain what changed and why it matters. + When a blocker or confusing implementation detail is solved, record the + symptom, cause, and fix in the relevant durable log or plan note. - Keep `docs/logs/long-term-thinking-log.md` in sync with any durable intent or success-criteria changes. - Keep `docs/runbooks/` aligned with the current CLI and server behavior. - For a new worktree, run `scripts/init.sh ` first. `scripts/bootstrap-worktree.sh` is only a compatibility wrapper. `scripts/init.sh` creates the worktree, downloads dependencies, builds local binaries, writes a sourceable env file, and can start `harnessd` in tmux when requested. diff --git a/cmd/harnessd/bootstrap_helpers.go b/cmd/harnessd/bootstrap_helpers.go index ef84380..eee6edc 100644 --- a/cmd/harnessd/bootstrap_helpers.go +++ b/cmd/harnessd/bootstrap_helpers.go @@ -25,6 +25,7 @@ import ( istore "go-agent-harness/internal/store" "go-agent-harness/internal/subagents" "go-agent-harness/internal/trigger" + scriptworkflow "go-agent-harness/internal/workflow" "go-agent-harness/internal/workflows" ) @@ -367,6 +368,7 @@ type serverBootstrapOptions struct { subagentManager subagents.Manager checkpoints *checkpoints.Service workflows *workflows.Engine + scriptWorkflows scriptworkflow.SourceService networks *networks.Engine providerRegistry *catalog.ProviderRegistry runStore istore.Store @@ -386,6 +388,7 @@ func buildServerOptions(opts serverBootstrapOptions) server.ServerOptions { SubagentManager: opts.subagentManager, Checkpoints: opts.checkpoints, Workflows: opts.workflows, + ScriptWorkflows: opts.scriptWorkflows, Networks: opts.networks, ProviderRegistry: opts.providerRegistry, Store: opts.runStore, diff --git a/cmd/harnessd/bootstrap_helpers_test.go b/cmd/harnessd/bootstrap_helpers_test.go index e3cd57f..85d844e 100644 --- a/cmd/harnessd/bootstrap_helpers_test.go +++ b/cmd/harnessd/bootstrap_helpers_test.go @@ -7,11 +7,15 @@ import ( "path/filepath" "strings" "testing" + "time" "go-agent-harness/internal/harness" + htools "go-agent-harness/internal/harness/tools" "go-agent-harness/internal/provider/catalog" openai "go-agent-harness/internal/provider/openai" "go-agent-harness/internal/relay" + "go-agent-harness/internal/subagents" + scriptworkflow "go-agent-harness/internal/workflow" ) func TestBuildCatalogBootstrapFallsBackToWorkspaceCatalog(t *testing.T) { @@ -144,6 +148,7 @@ func TestBuildServerOptionsForwardsBootstrapRuntime(t *testing.T) { opts := buildServerOptions(serverBootstrapOptions{ runner: runner, modelCatalog: cat, + scriptWorkflows: &scriptWorkflowServiceRef{}, providerRegistry: catalog.NewProviderRegistryWithEnv(cat, func(string) string { return "" }), triggers: runtime, relayWorkerStore: relayStore, @@ -161,6 +166,9 @@ func TestBuildServerOptionsForwardsBootstrapRuntime(t *testing.T) { if opts.ProviderRegistry == nil { t.Fatal("expected provider registry") } + if opts.ScriptWorkflows == nil { + t.Fatal("expected script workflows") + } if opts.Validators == nil { t.Fatal("expected validators") } @@ -178,6 +186,235 @@ func TestBuildServerOptionsForwardsBootstrapRuntime(t *testing.T) { } } +func TestScriptWorkflowServiceRefDelegatesAfterBinding(t *testing.T) { + t.Parallel() + + engine := scriptworkflow.NewEngine(scriptworkflow.EngineOptions{Subagents: sourceNoopSubagentsForHarnessd{}}) + manager, err := scriptworkflow.NewSourceManager(scriptworkflow.SourceManagerOptions{ + Engine: engine, + CacheDir: t.TempDir(), + }) + if err != nil { + t.Fatalf("NewSourceManager: %v", err) + } + ref := &scriptWorkflowServiceRef{} + if _, err := ref.Start(t.Context(), "missing", nil); err == nil { + t.Fatal("expected unbound ref to reject start") + } + if err := ref.Reload(t.Context()); err != nil { + t.Fatalf("unbound reload should be a no-op: %v", err) + } + ref.Set(manager) + if got := ref.List(); got == nil { + t.Fatal("expected delegated list to return a non-nil slice") + } + if err := ref.Reload(t.Context()); err != nil { + t.Fatalf("bound reload: %v", err) + } +} + +func TestScriptWorkflowServiceRefDelegatesAllMethods(t *testing.T) { + t.Parallel() + + svc := &fakeScriptWorkflowServiceForHarnessd{ + run: &scriptworkflow.Run{ID: "wf_1", WorkflowName: "daily-review", Status: scriptworkflow.RunStatusCompleted}, + events: []scriptworkflow.Event{{ + RunID: "wf_1", + Type: scriptworkflow.EventWorkflowCompleted, + }}, + } + ref := &scriptWorkflowServiceRef{} + ref.Set(svc) + + if _, err := ref.Resume(t.Context(), "wf_1", map[string]any{"resume": true}); err != nil { + t.Fatalf("Resume: %v", err) + } + if _, err := ref.GetRun("wf_1"); err != nil { + t.Fatalf("GetRun: %v", err) + } + history, stream, cancel, err := ref.Subscribe("wf_1") + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + cancel() + if len(history) != 1 || stream == nil { + t.Fatalf("Subscribe history=%d stream nil=%v", len(history), stream == nil) + } + if _, _, err := ref.Wait(t.Context(), "wf_1"); err != nil { + t.Fatalf("Wait: %v", err) + } + bundle, err := ref.CreateWorkflow(t.Context(), scriptworkflow.CreateWorkflowRequest{Name: "daily-review"}) + if err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + if bundle.Manifest.Name != "daily-review" { + t.Fatalf("bundle name = %q", bundle.Manifest.Name) + } +} + +func TestScriptSubagentAdapterMapsRequestsAndResults(t *testing.T) { + t.Parallel() + + manager := &fakeSubagentManagerForHarnessd{ + item: subagents.Subagent{ + ID: "subagent_1", + RunID: "run_1", + Status: harness.RunStatusCompleted, + Output: "done", + }, + } + adapter := scriptSubagentAdapter{manager: manager} + result, err := adapter.Create(t.Context(), scriptworkflow.SubagentRequest{ + Prompt: "inspect", + Model: "gpt-5-nano", + Provider: "openai", + Profile: "reviewer", + AllowedTools: []string{"read"}, + Isolation: "worktree", + CleanupPolicy: "preserve", + MaxSteps: 5, + MaxCostUSD: 0.5, + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + if result.ID != "subagent_1" || result.Output != "done" { + t.Fatalf("unexpected result: %#v", result) + } + if manager.req.Prompt != "inspect" || manager.req.ProviderName != "openai" || manager.req.ProfileName != "reviewer" { + t.Fatalf("unexpected mapped request: %#v", manager.req) + } + if got := manager.req.AllowedTools; len(got) != 1 || got[0] != "read" { + t.Fatalf("allowed tools = %#v", got) + } + if manager.req.Isolation != subagents.IsolationWorktree || manager.req.CleanupPolicy != subagents.CleanupPreserve { + t.Fatalf("isolation=%q cleanup=%q", manager.req.Isolation, manager.req.CleanupPolicy) + } + + got, err := adapter.Get(t.Context(), "subagent_1") + if err != nil { + t.Fatalf("Get: %v", err) + } + if got.ID != "subagent_1" || got.Status != string(harness.RunStatusCompleted) { + t.Fatalf("unexpected get result: %#v", got) + } +} + +func TestWorkflowQuestionResponderUsesAskBroker(t *testing.T) { + t.Parallel() + + broker := &fakeAskUserQuestionBrokerForHarnessd{} + responder := workflowQuestionResponder{broker: broker, timeout: 2 * time.Second} + answer, err := responder.AskWorkflowQuestion(t.Context(), scriptworkflow.QuestionRequest{ + RunID: "wf_1", + CallID: "call_1", + Prompt: "Continue?", + Choices: []scriptworkflow.QuestionOption{{ + Label: "Continue", + Description: "Keep going.", + }}, + }) + if err != nil { + t.Fatalf("AskWorkflowQuestion: %v", err) + } + if answer != "Continue" { + t.Fatalf("answer = %#v", answer) + } + if broker.req.RunID != "wf_1" || broker.req.CallID != "call_1" || broker.req.Timeout != 2*time.Second { + t.Fatalf("broker request = %#v", broker.req) + } + if len(broker.req.Questions) != 1 || broker.req.Questions[0].Question != "Continue?" { + t.Fatalf("questions = %#v", broker.req.Questions) + } +} + +type fakeScriptWorkflowServiceForHarnessd struct { + run *scriptworkflow.Run + events []scriptworkflow.Event +} + +func (f *fakeScriptWorkflowServiceForHarnessd) List() []scriptworkflow.Meta { + return []scriptworkflow.Meta{{Name: "daily-review"}} +} + +func (f *fakeScriptWorkflowServiceForHarnessd) Start(context.Context, string, any) (*scriptworkflow.Run, error) { + return f.run, nil +} + +func (f *fakeScriptWorkflowServiceForHarnessd) Resume(context.Context, string, any) (*scriptworkflow.Run, error) { + return f.run, nil +} + +func (f *fakeScriptWorkflowServiceForHarnessd) GetRun(string) (*scriptworkflow.Run, error) { + return f.run, nil +} + +func (f *fakeScriptWorkflowServiceForHarnessd) Subscribe(string) ([]scriptworkflow.Event, <-chan scriptworkflow.Event, func(), error) { + ch := make(chan scriptworkflow.Event) + return f.events, ch, func() { close(ch) }, nil +} + +func (f *fakeScriptWorkflowServiceForHarnessd) Wait(context.Context, string) (*scriptworkflow.Run, []scriptworkflow.Event, error) { + return f.run, f.events, nil +} + +func (f *fakeScriptWorkflowServiceForHarnessd) CreateWorkflow(_ context.Context, req scriptworkflow.CreateWorkflowRequest) (*scriptworkflow.SourceBundle, error) { + return &scriptworkflow.SourceBundle{Manifest: scriptworkflow.SourceBundleManifest{Name: req.Name}}, nil +} + +type fakeSubagentManagerForHarnessd struct { + req subagents.Request + item subagents.Subagent +} + +func (f *fakeSubagentManagerForHarnessd) Create(_ context.Context, req subagents.Request) (subagents.Subagent, error) { + f.req = req + return f.item, nil +} + +func (f *fakeSubagentManagerForHarnessd) Get(context.Context, string) (subagents.Subagent, error) { + return f.item, nil +} + +func (f *fakeSubagentManagerForHarnessd) List(context.Context) ([]subagents.Subagent, error) { + return nil, nil +} + +func (f *fakeSubagentManagerForHarnessd) Delete(context.Context, string) error { + return nil +} + +func (f *fakeSubagentManagerForHarnessd) Cancel(context.Context, string) error { + return nil +} + +type fakeAskUserQuestionBrokerForHarnessd struct { + req htools.AskUserQuestionRequest +} + +func (f *fakeAskUserQuestionBrokerForHarnessd) Ask(_ context.Context, req htools.AskUserQuestionRequest) (map[string]string, time.Time, error) { + f.req = req + return map[string]string{req.Questions[0].Question: "Continue"}, time.Unix(1, 0), nil +} + +func (f *fakeAskUserQuestionBrokerForHarnessd) Pending(string) (htools.AskUserQuestionPending, bool) { + return htools.AskUserQuestionPending{}, false +} + +func (f *fakeAskUserQuestionBrokerForHarnessd) Submit(string, map[string]string) error { + return nil +} + +type sourceNoopSubagentsForHarnessd struct{} + +func (sourceNoopSubagentsForHarnessd) Create(context.Context, scriptworkflow.SubagentRequest) (scriptworkflow.SubagentResult, error) { + return scriptworkflow.SubagentResult{ID: "subagent_1", Status: "completed"}, nil +} + +func (sourceNoopSubagentsForHarnessd) Get(context.Context, string) (scriptworkflow.SubagentResult, error) { + return scriptworkflow.SubagentResult{ID: "subagent_1", Status: "completed"}, nil +} + func TestBuildPersistenceBootstrapInitializesStoresAndCleaner(t *testing.T) { t.Parallel() diff --git a/cmd/harnessd/main.go b/cmd/harnessd/main.go index 98e8408..f6ccb4a 100644 --- a/cmd/harnessd/main.go +++ b/cmd/harnessd/main.go @@ -660,7 +660,16 @@ func runWithSignalsWithDeps(sig <-chan os.Signal, getenv func(string) string, ne approvalBroker := harness.NewCheckpointApprovalBroker(checkpointService) activations := harness.NewActivationTracker() msgSummarizer := &lazySummarizer{} + scriptWorkflowRef := &scriptWorkflowServiceRef{} promptBehaviorsDir, promptTalentsDir := promptEngine.ExtensionDirs() + globalWorkflowsDir := filepath.Join(globalDir, "workflows") + workspaceWorkflowsDir := filepath.Join(workspace, ".go-harness", "workflows") + globalSkillsDir := filepath.Join(globalDir, "skills") + workspaceSkillsDir := filepath.Join(workspace, ".go-harness", "skills") + goWorkflowCacheDir := strings.TrimSpace(getenv("HARNESS_GO_WORKFLOW_CACHE_DIR")) + if goWorkflowCacheDir == "" { + goWorkflowCacheDir = filepath.Join(workspace, ".harness", "workflow-cache") + } baseRegistryOptions := harness.DefaultRegistryOptions{ ApprovalMode: approvalMode, Policy: nil, @@ -684,6 +693,7 @@ func runWithSignalsWithDeps(sig <-chan os.Signal, getenv func(string) string, ne TalentsDir: promptTalentsDir, }, ScriptToolsDir: filepath.Join(globalDir, "tools"), + WorkflowService: scriptWorkflowRef, ConversationStore: convStore, MessageSummarizer: msgSummarizer, MCPRegistry: mcpRegistry, @@ -766,19 +776,31 @@ func runWithSignalsWithDeps(sig <-chan os.Signal, getenv func(string) string, ne log.Printf("watcher: skill reload error: %v", err) return err } + if err := scriptWorkflowRef.Reload(context.Background()); err != nil { + log.Printf("watcher: workflow reload error: %v", err) + return err + } log.Printf("watcher: skills reloaded (%d skill(s))", len(skillRegistry.List())) return nil } - globalSkillsDir := filepath.Join(globalDir, "skills") - workspaceSkillsDir := filepath.Join(workspace, ".go-harness", "skills") + reloadWorkflows := func() error { + if err := scriptWorkflowRef.Reload(context.Background()); err != nil { + log.Printf("watcher: workflow reload error: %v", err) + return err + } + log.Printf("watcher: script workflows reloaded") + return nil + } + w.Watch(watcher.WatchedDir{Path: globalWorkflowsDir, Reload: reloadWorkflows}) + w.Watch(watcher.WatchedDir{Path: workspaceWorkflowsDir, Reload: reloadWorkflows}) w.Watch(watcher.WatchedDir{Path: globalSkillsDir, Reload: reloadSkills}) w.Watch(watcher.WatchedDir{Path: workspaceSkillsDir, Reload: reloadSkills}) go w.Start(watchCtx) - log.Printf("hot-reload watcher started (interval: %s, dirs: %s, %s)", - pollInterval, globalSkillsDir, workspaceSkillsDir) + log.Printf("hot-reload watcher started (interval: %s, dirs: %s, %s, %s, %s)", + pollInterval, globalWorkflowsDir, workspaceWorkflowsDir, globalSkillsDir, workspaceSkillsDir) } // Build the Skills SkillManager only when skillLister is a *skillListerAdapter @@ -803,6 +825,10 @@ func runWithSignalsWithDeps(sig <-chan os.Signal, getenv func(string) string, ne checkpointService: checkpointService, workflowDefinitions: workflowDefinitions, workflowStore: workflowStore, + goWorkflowDirs: []string{globalWorkflowsDir, workspaceWorkflowsDir}, + goWorkflowSkillDirs: []string{globalSkillsDir, workspaceSkillsDir}, + goWorkflowCacheDir: goWorkflowCacheDir, + scriptWorkflowRef: scriptWorkflowRef, networkDefinitions: networkDefinitions, skillLister: skillLister, baseRegistryOptions: baseRegistryOptions, @@ -819,6 +845,8 @@ func runWithSignalsWithDeps(sig <-chan os.Signal, getenv func(string) string, ne subagentBaseRef: subagentBaseRef, subagentWorktreeRoot: subagentWorktreeRoot, subagentConfigTOML: subagentConfigTOML, + askUserBroker: askUserBroker, + askUserTimeout: time.Duration(askUserTimeoutSeconds) * time.Second, }) if err != nil { return err diff --git a/cmd/harnessd/runtime_container.go b/cmd/harnessd/runtime_container.go index 03e4abc..a808866 100644 --- a/cmd/harnessd/runtime_container.go +++ b/cmd/harnessd/runtime_container.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "net/http" "strings" @@ -16,6 +17,7 @@ import ( "go-agent-harness/internal/server" istore "go-agent-harness/internal/store" "go-agent-harness/internal/subagents" + scriptworkflow "go-agent-harness/internal/workflow" "go-agent-harness/internal/workflows" ) @@ -61,6 +63,10 @@ type httpRuntimeOptions struct { checkpointService *checkpoints.Service workflowDefinitions []workflows.Definition workflowStore workflows.Store + goWorkflowDirs []string + goWorkflowSkillDirs []string + goWorkflowCacheDir string + scriptWorkflowRef *scriptWorkflowServiceRef networkDefinitions []networks.Definition skillLister htools.SkillLister baseRegistryOptions harness.DefaultRegistryOptions @@ -77,6 +83,8 @@ type httpRuntimeOptions struct { subagentBaseRef string subagentWorktreeRoot string subagentConfigTOML string + askUserBroker htools.AskUserQuestionBroker + askUserTimeout time.Duration } type httpRuntime struct { @@ -119,6 +127,29 @@ func buildHTTPRuntime(opts httpRuntimeOptions) (httpRuntime, error) { return httpRuntime{}, fmt.Errorf("create subagent manager: %w", err) } + scriptEngine := scriptworkflow.NewEngine(scriptworkflow.EngineOptions{ + Subagents: scriptSubagentAdapter{manager: subagentMgr}, + QuestionResponder: workflowQuestionResponder{ + broker: opts.askUserBroker, + timeout: opts.askUserTimeout, + }, + }) + sourceWorkflows, err := scriptworkflow.NewSourceManager(scriptworkflow.SourceManagerOptions{ + Engine: scriptEngine, + WorkflowDirs: opts.goWorkflowDirs, + SkillDirs: opts.goWorkflowSkillDirs, + CacheDir: opts.goWorkflowCacheDir, + }) + if err != nil { + return httpRuntime{}, fmt.Errorf("create script workflow manager: %w", err) + } + if err := sourceWorkflows.Load(context.Background()); err != nil { + return httpRuntime{}, fmt.Errorf("load script workflows: %w", err) + } + if opts.scriptWorkflowRef != nil { + opts.scriptWorkflowRef.Set(sourceWorkflows) + } + if opts.callbackStarter != nil { opts.callbackStarter.mu.Lock() opts.callbackStarter.runner = runner @@ -144,6 +175,7 @@ func buildHTTPRuntime(opts httpRuntimeOptions) (httpRuntime, error) { subagentManager: subagentMgr, checkpoints: opts.checkpointService, workflows: workflowEngine, + scriptWorkflows: sourceWorkflows, networks: networkEngine, providerRegistry: opts.providerRegistry, runStore: opts.runStore, diff --git a/cmd/harnessd/script_workflows.go b/cmd/harnessd/script_workflows.go new file mode 100644 index 0000000..a714097 --- /dev/null +++ b/cmd/harnessd/script_workflows.go @@ -0,0 +1,199 @@ +package main + +import ( + "context" + "fmt" + "sync" + "time" + + htools "go-agent-harness/internal/harness/tools" + "go-agent-harness/internal/subagents" + scriptworkflow "go-agent-harness/internal/workflow" +) + +type scriptWorkflowServiceRef struct { + mu sync.RWMutex + svc scriptworkflow.SourceService +} + +func (r *scriptWorkflowServiceRef) Set(svc scriptworkflow.SourceService) { + r.mu.Lock() + defer r.mu.Unlock() + r.svc = svc +} + +func (r *scriptWorkflowServiceRef) service() (scriptworkflow.SourceService, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if r.svc == nil { + return nil, fmt.Errorf("script workflow service is not configured") + } + return r.svc, nil +} + +func (r *scriptWorkflowServiceRef) Reload(ctx context.Context) error { + r.mu.RLock() + svc := r.svc + r.mu.RUnlock() + if svc == nil { + return nil + } + reloader, ok := svc.(interface { + Load(context.Context) error + }) + if !ok { + return nil + } + return reloader.Load(ctx) +} + +func (r *scriptWorkflowServiceRef) List() []scriptworkflow.Meta { + svc, err := r.service() + if err != nil { + return nil + } + return svc.List() +} + +func (r *scriptWorkflowServiceRef) Start(ctx context.Context, name string, args any) (*scriptworkflow.Run, error) { + svc, err := r.service() + if err != nil { + return nil, err + } + return svc.Start(ctx, name, args) +} + +func (r *scriptWorkflowServiceRef) Resume(ctx context.Context, runID string, args any) (*scriptworkflow.Run, error) { + svc, err := r.service() + if err != nil { + return nil, err + } + return svc.Resume(ctx, runID, args) +} + +func (r *scriptWorkflowServiceRef) GetRun(runID string) (*scriptworkflow.Run, error) { + svc, err := r.service() + if err != nil { + return nil, err + } + return svc.GetRun(runID) +} + +func (r *scriptWorkflowServiceRef) Subscribe(runID string) ([]scriptworkflow.Event, <-chan scriptworkflow.Event, func(), error) { + svc, err := r.service() + if err != nil { + return nil, nil, nil, err + } + return svc.Subscribe(runID) +} + +func (r *scriptWorkflowServiceRef) Wait(ctx context.Context, runID string) (*scriptworkflow.Run, []scriptworkflow.Event, error) { + svc, err := r.service() + if err != nil { + return nil, nil, err + } + return svc.Wait(ctx, runID) +} + +func (r *scriptWorkflowServiceRef) CreateWorkflow(ctx context.Context, req scriptworkflow.CreateWorkflowRequest) (*scriptworkflow.SourceBundle, error) { + svc, err := r.service() + if err != nil { + return nil, err + } + return svc.CreateWorkflow(ctx, req) +} + +type scriptSubagentAdapter struct { + manager subagents.Manager +} + +func (a scriptSubagentAdapter) Create(ctx context.Context, req scriptworkflow.SubagentRequest) (scriptworkflow.SubagentResult, error) { + if a.manager == nil { + return scriptworkflow.SubagentResult{}, fmt.Errorf("subagent manager is not configured") + } + item, err := a.manager.Create(ctx, subagents.Request{ + Prompt: req.Prompt, + Model: req.Model, + ProviderName: req.Provider, + AllowedTools: append([]string(nil), req.AllowedTools...), + ProfileName: req.Profile, + Isolation: subagents.IsolationMode(req.Isolation), + CleanupPolicy: subagents.CleanupPolicy(req.CleanupPolicy), + MaxSteps: req.MaxSteps, + MaxCostUSD: req.MaxCostUSD, + }) + if err != nil { + return scriptworkflow.SubagentResult{}, err + } + return subagentResultFrom(item), nil +} + +func (a scriptSubagentAdapter) Get(ctx context.Context, id string) (scriptworkflow.SubagentResult, error) { + if a.manager == nil { + return scriptworkflow.SubagentResult{}, fmt.Errorf("subagent manager is not configured") + } + item, err := a.manager.Get(ctx, id) + if err != nil { + return scriptworkflow.SubagentResult{}, err + } + return subagentResultFrom(item), nil +} + +func subagentResultFrom(item subagents.Subagent) scriptworkflow.SubagentResult { + return scriptworkflow.SubagentResult{ + ID: item.ID, + Status: string(item.Status), + Output: item.Output, + Error: item.Error, + } +} + +type workflowQuestionResponder struct { + broker htools.AskUserQuestionBroker + timeout time.Duration +} + +func (r workflowQuestionResponder) AskWorkflowQuestion(ctx context.Context, req scriptworkflow.QuestionRequest) (any, error) { + if r.broker == nil { + return nil, fmt.Errorf("workflow question broker is not configured") + } + options := make([]htools.AskUserQuestionOption, 0, len(req.Choices)) + for _, choice := range req.Choices { + if len(options) >= 4 { + break + } + label := choice.Label + if label == "" { + continue + } + desc := choice.Description + if desc == "" { + desc = choice.Label + } + options = append(options, htools.AskUserQuestionOption{Label: label, Description: desc}) + } + if len(options) == 0 { + options = append(options, + htools.AskUserQuestionOption{Label: "Continue", Description: "Continue the workflow."}, + htools.AskUserQuestionOption{Label: "Stop", Description: "Stop and report the blocker."}, + ) + } + if len(options) == 1 { + options = append(options, htools.AskUserQuestionOption{Label: "Other", Description: "Use another answer."}) + } + answers, _, err := r.broker.Ask(ctx, htools.AskUserQuestionRequest{ + RunID: req.RunID, + CallID: req.CallID, + Questions: []htools.AskUserQuestion{{ + Question: req.Prompt, + Header: "Workflow", + Options: options, + MultiSelect: false, + }}, + Timeout: r.timeout, + }) + if err != nil { + return nil, err + } + return answers[req.Prompt], nil +} diff --git a/docs/logs/long-term-thinking-log.md b/docs/logs/long-term-thinking-log.md index 2637065..0bae066 100644 --- a/docs/logs/long-term-thinking-log.md +++ b/docs/logs/long-term-thinking-log.md @@ -1,5 +1,27 @@ # Long-Term Thinking Log +## 2026-06-27 (Go-Authored Custom Workflows) + +- Command intent: Implement custom workflows that agents can create, save, run, and monitor as Go source without restarting `harnessd`. +- User intent: Bring go-code closer to Claude-style custom workflows while keeping workflows transferable through skills and usable by parent agents through feedback. +- Success definition: + - Agents can create validated Go workflow bundles through `create_workflow`. + - Agents can run workflows through `run_workflow` and receive structured feedback events. + - Workflow bundles are discovered from global, workspace, and skill directories. + - Watched workflow and skill directories can reload discovered bundles without a `harnessd` restart. + - `/v1/script-workflows` is backed by a real runtime in `harnessd`. + - Focused tests cover creation, discovery, feedback, question events, child-process failure handling, subagent RPC forwarding, SSE history, and wiring. +- Non-goals: + - Replacing the existing YAML `/v1/workflows` runtime. + - Full Claude plugin parity for commands, hooks, agents, and MCP manifests in this slice. +- Guardrails/constraints: + - Treat Go workflow code as trusted local automation, similar to script tools. + - Keep generated workflow execution in a child process so `harnessd` stays alive. +- Solved struggle: Go plugins looked tempting for dynamic loading, but child-process binaries are safer for hot reload, cancellation, and repeated agent-generated code. +- Solved struggle: The embedded description regression failed because new markdown tool descriptions must be added to both explicit inventories in `embed_test.go`; the fix was to register `create_workflow` and `run_workflow` in both lists. +- Solved struggle: Directory discovery alone did not satisfy no-restart hot loading for manually saved bundles; the fix was to route the existing watcher through `scriptWorkflowServiceRef.Reload` for workflow dirs and skill dirs. +- Solved struggle: The full regression later failed the no-zero-coverage rule even though total coverage was above threshold; the fix was to add behavior tests for the new harnessd adapters, nested workflow/stderr/resume paths, and uncovered TUI editor helpers instead of weakening coveragegate. + ## 2026-06-28 (Go Relay PR #689 Review Repair) - Command intent: Fix the PR that implements the Go Relay epic by addressing the blocker review findings and pushing the repaired branch back to the existing PR. diff --git a/docs/plans/2026-06-27-go-source-workflows-plan.md b/docs/plans/2026-06-27-go-source-workflows-plan.md new file mode 100644 index 0000000..43fb99c --- /dev/null +++ b/docs/plans/2026-06-27-go-source-workflows-plan.md @@ -0,0 +1,27 @@ +# Plan: Go-Authored Custom Workflows + +## Intent + +- Command intent: let agents create, save, hot-load, run, and monitor custom workflows written in Go. +- User intent: match the practical Claude workflow pattern while keeping go-code local-first and reusable through skills. +- Success definition: workflow bundles compile without restarting `harnessd`, can be created and run through tools, emit feedback to the parent agent/API stream, and can be bundled under skills. + +## Implementation Shape + +- Workflow bundles live under `.go-harness/workflows//` or under a skill at `.go-harness/skills//workflows//`. +- Each bundle contains `workflow.json` and Go source using `go-agent-harness/pkg/workflowsdk`. +- The host compiles workflow source into a cache and executes it as a child process using a JSONL protocol. +- The child process calls back into the host for agent runs, nested workflows, phases, logs, feedback, and questions. +- Existing YAML `/v1/workflows` behavior is left unchanged; Go-authored workflows use the existing `/v1/script-workflows` route family. + +## Solved Struggle + +- Symptom: Go workflows need to be created and run without stopping `harnessd`, but Go plugins are brittle and cannot unload cleanly. +- Cause: Go plugin ABI compatibility and process lifetime do not fit agent-generated code that should be hot-loaded repeatedly. +- Fix: compile workflows as child binaries and bridge to the host over JSONL stdio; this preserves hot loading, cancellation, bounded diagnostics, and host-owned subagent execution. + +## Verification + +- Added tests for dynamic workflow creation/build/run, skill-bundled discovery, feedback propagation, questions, create/run tools, SSE feedback history, subagent RPC forwarding, process/protocol failure handling, harnessd script workflow wiring, and coveragegate zero-function gaps. +- Verified with focused package tests and `./scripts/test-regression.sh`. +- Final regression evidence: `coveragegate: PASS (total=84.4%, min=80.0%, zero-functions=0)` and `[regression] PASS`. diff --git a/docs/plans/INDEX.md b/docs/plans/INDEX.md index ba8b228..5c6a911 100644 --- a/docs/plans/INDEX.md +++ b/docs/plans/INDEX.md @@ -5,6 +5,7 @@ - `active-plan.md`: Current active plan tracker. - `2026-06-26-adapter-first-eval-harness-plan.md`: Implemented adapter-first Terminal-Bench eval harness hardening plan; full regression and real-provider smoke baseline acceptance are green. - `2026-06-24-harness-reliability-plan.md`: Active 15-slice TDD plan (T01–T15) hardening `harnessd` against long-session reliability failures from the 2026-06-24 audit; tracked via `.context/harness-reliability/tracker.md`. +- `2026-06-27-go-source-workflows-plan.md`: Completed implementation plan for Go-authored custom workflow bundles, child-process execution, feedback events, and skill-bundled transfer. - `2026-04-05-orchestration-program-plan.md`: Umbrella staged architecture plan for runtime container, checkpoints, workflows, memory layering, and agent networks with documentation/TDD guardrails. - `2026-04-05-runtime-container-spec.md`: Stage 1 spec for the internal runtime composition root extraction in `cmd/harnessd`. - `2026-04-05-durable-checkpoints-spec.md`: Stage 2 spec for persisted human-in-the-loop checkpoints and compatibility shims. diff --git a/internal/harness/tools/deferred/workflow.go b/internal/harness/tools/deferred/workflow.go new file mode 100644 index 0000000..2911422 --- /dev/null +++ b/internal/harness/tools/deferred/workflow.go @@ -0,0 +1,162 @@ +package deferred + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + tools "go-agent-harness/internal/harness/tools" + "go-agent-harness/internal/harness/tools/descriptions" + "go-agent-harness/internal/workflow" +) + +func CreateWorkflowTool(manager workflow.SourceService) tools.Tool { + def := tools.Definition{ + Name: "create_workflow", + Description: descriptions.Load("create_workflow"), + Action: tools.ActionWrite, + Mutating: true, + ParallelSafe: false, + Tier: tools.TierDeferred, + Tags: []string{"workflow", "go", "automation", "create"}, + Parameters: map[string]any{ + "type": "object", + "properties": map[string]any{ + "name": map[string]any{"type": "string", "description": "Kebab-case workflow name."}, + "description": map[string]any{"type": "string", "description": "What the workflow does."}, + "when_to_use": map[string]any{"type": "string", "description": "When an agent should use this workflow."}, + "source": map[string]any{"type": "string", "description": "Full Go source for package main using go-agent-harness/pkg/workflowsdk."}, + "scope": map[string]any{"type": "string", "enum": []string{"workspace", "global", "skill"}, "description": "Where to save the workflow. Defaults to workspace."}, + "skill": map[string]any{"type": "string", "description": "Skill name when scope=skill."}, + "timeout_seconds": map[string]any{"type": "integer", "description": "Optional per-run timeout."}, + "overwrite": map[string]any{"type": "boolean", "description": "Replace an existing workflow with the same name."}, + "args_schema": map[string]any{"type": "object", "description": "Optional JSON-schema-like argument schema."}, + }, + "required": []string{"name", "description", "source"}, + "additionalProperties": false, + }, + } + handler := func(ctx context.Context, raw json.RawMessage) (string, error) { + if manager == nil { + return "", fmt.Errorf("create_workflow: workflow service is not configured") + } + var args workflow.CreateWorkflowRequest + if err := json.Unmarshal(raw, &args); err != nil { + return "", fmt.Errorf("parse create_workflow args: %w", err) + } + bundle, err := manager.CreateWorkflow(ctx, args) + if err != nil { + return "", err + } + return tools.MarshalToolResult(map[string]any{ + "status": "created", + "name": bundle.Manifest.Name, + "path": bundle.Dir, + "hash": bundle.Hash, + "scope": bundle.Scope, + }) + } + return tools.Tool{Definition: def, Handler: handler} +} + +func RunWorkflowTool(manager workflow.SourceService) tools.Tool { + def := tools.Definition{ + Name: "run_workflow", + Description: descriptions.Load("run_workflow"), + Action: tools.ActionExecute, + Mutating: true, + ParallelSafe: false, + Tier: tools.TierDeferred, + Tags: []string{"workflow", "go", "automation", "run"}, + Parameters: map[string]any{ + "type": "object", + "properties": map[string]any{ + "name": map[string]any{"type": "string", "description": "Workflow name to run."}, + "args": map[string]any{"type": "object", "description": "Workflow arguments."}, + "wait": map[string]any{"type": "boolean", "description": "Wait for terminal result. Defaults to true."}, + "timeout_seconds": map[string]any{"type": "integer", "description": "Tool wait timeout."}, + "resume_run_id": map[string]any{"type": "string", "description": "Failed workflow run id to resume."}, + }, + "required": []string{"name"}, + "additionalProperties": false, + }, + } + handler := func(ctx context.Context, raw json.RawMessage) (string, error) { + if manager == nil { + return "", fmt.Errorf("run_workflow: workflow service is not configured") + } + var args struct { + Name string `json:"name"` + Args map[string]any `json:"args"` + Wait *bool `json:"wait"` + TimeoutSeconds int `json:"timeout_seconds"` + ResumeRunID string `json:"resume_run_id"` + } + if err := json.Unmarshal(raw, &args); err != nil { + return "", fmt.Errorf("parse run_workflow args: %w", err) + } + name := strings.TrimSpace(args.Name) + if name == "" && strings.TrimSpace(args.ResumeRunID) == "" { + return "", fmt.Errorf("run_workflow: name is required") + } + wait := true + if args.Wait != nil { + wait = *args.Wait + } + runArgs := map[string]any{} + if args.Args != nil { + runArgs = args.Args + } + var run *workflow.Run + var err error + if strings.TrimSpace(args.ResumeRunID) != "" { + run, err = manager.Resume(ctx, strings.TrimSpace(args.ResumeRunID), runArgs) + } else { + run, err = manager.Start(ctx, name, runArgs) + } + if err != nil { + return "", err + } + events := []workflow.Event{} + if wait { + waitCtx := ctx + cancel := func() {} + if args.TimeoutSeconds > 0 { + waitCtx, cancel = context.WithTimeout(ctx, time.Duration(args.TimeoutSeconds)*time.Second) + } + defer cancel() + run, events, err = manager.Wait(waitCtx, run.ID) + if err != nil { + return "", err + } + } + return tools.MarshalToolResult(map[string]any{ + "run_id": run.ID, + "workflow_name": run.WorkflowName, + "status": run.Status, + "result_json": run.ResultJSON, + "error": run.Error, + "feedback": workflowFeedback(events), + }) + } + return tools.Tool{Definition: def, Handler: handler} +} + +func workflowFeedback(events []workflow.Event) []map[string]any { + out := []map[string]any{} + for _, ev := range events { + switch ev.Type { + case workflow.EventWorkflowFeedback, workflow.EventWorkflowFinding, workflow.EventWorkflowWarning, workflow.EventWorkflowQuestion, workflow.EventWorkflowLog, workflow.EventWorkflowPhaseStarted: + item := map[string]any{ + "seq": ev.Seq, + "type": ev.Type, + "payload": ev.Payload, + "timestamp": ev.Timestamp, + } + out = append(out, item) + } + } + return out +} diff --git a/internal/harness/tools/deferred/workflow_test.go b/internal/harness/tools/deferred/workflow_test.go new file mode 100644 index 0000000..3b0df22 --- /dev/null +++ b/internal/harness/tools/deferred/workflow_test.go @@ -0,0 +1,90 @@ +package deferred + +import ( + "context" + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "go-agent-harness/internal/workflow" +) + +type fakeWorkflowService struct { + created workflow.CreateWorkflowRequest + run *workflow.Run + events []workflow.Event +} + +func (f *fakeWorkflowService) List() []workflow.Meta { return nil } + +func (f *fakeWorkflowService) Start(_ context.Context, name string, _ any) (*workflow.Run, error) { + f.run = &workflow.Run{ID: "wf_1", WorkflowName: name, Status: workflow.RunStatusRunning} + return f.run, nil +} + +func (f *fakeWorkflowService) Resume(context.Context, string, any) (*workflow.Run, error) { + return nil, fmt.Errorf("not implemented") +} + +func (f *fakeWorkflowService) GetRun(string) (*workflow.Run, error) { return f.run, nil } + +func (f *fakeWorkflowService) Subscribe(string) ([]workflow.Event, <-chan workflow.Event, func(), error) { + return nil, nil, func() {}, nil +} + +func (f *fakeWorkflowService) Wait(context.Context, string) (*workflow.Run, []workflow.Event, error) { + f.run.Status = workflow.RunStatusCompleted + f.run.ResultJSON = `{"ok":true}` + return f.run, f.events, nil +} + +func (f *fakeWorkflowService) CreateWorkflow(_ context.Context, req workflow.CreateWorkflowRequest) (*workflow.SourceBundle, error) { + f.created = req + return &workflow.SourceBundle{ + Manifest: workflow.SourceBundleManifest{Name: req.Name, Description: req.Description}, + Dir: "/tmp/workflows/" + req.Name, + Hash: "abc123", + Scope: req.Scope, + }, nil +} + +func TestCreateWorkflowToolCreatesBundle(t *testing.T) { + fake := &fakeWorkflowService{} + tool := CreateWorkflowTool(fake) + out, err := tool.Handler(context.Background(), json.RawMessage(`{ + "name": "daily-review", + "description": "Review the day.", + "source": "package main\nfunc main(){}", + "scope": "workspace" + }`)) + require.NoError(t, err) + require.Contains(t, out, `"status":"created"`) + require.Equal(t, "daily-review", fake.created.Name) + require.Equal(t, "workspace", fake.created.Scope) +} + +func TestRunWorkflowToolReturnsFeedbackWhenWaiting(t *testing.T) { + fake := &fakeWorkflowService{ + events: []workflow.Event{{ + Seq: 1, + RunID: "wf_1", + Type: workflow.EventWorkflowFinding, + Payload: map[string]any{"kind": "finding", "message": "found issue"}, + Timestamp: time.Now().UTC(), + }}, + } + tool := RunWorkflowTool(fake) + out, err := tool.Handler(context.Background(), json.RawMessage(`{ + "name": "daily-review", + "args": {"target": "repo"}, + "wait": true + }`)) + require.NoError(t, err) + require.Contains(t, out, `"run_id":"wf_1"`) + require.Contains(t, out, `"status":"completed"`) + require.Contains(t, out, `"workflow.finding"`) + require.Contains(t, out, `"found issue"`) +} diff --git a/internal/harness/tools/descriptions/create_workflow.md b/internal/harness/tools/descriptions/create_workflow.md new file mode 100644 index 0000000..08a8274 --- /dev/null +++ b/internal/harness/tools/descriptions/create_workflow.md @@ -0,0 +1,20 @@ +Create a reusable Go-authored workflow and make it available immediately. + +The workflow is saved as a bundle containing `workflow.json` and `main.go`. +The Go source must be `package main` and should use: + +```go +import sdk "go-agent-harness/pkg/workflowsdk" +``` + +Call `sdk.Main(func(ctx *sdk.Context) (any, error) { ... })` from `main`. +Use `ctx.Agent`, `ctx.Phase`, `ctx.Log`, `ctx.Feedback`, `ctx.Question`, and +`ctx.Workflow` to coordinate work through the host harness. + +Scopes: +- `workspace`: save under the current workspace `.go-harness/workflows` +- `global`: save under the user-global workflows directory +- `skill`: save under a skill directory so the workflow transfers with the skill + +The workflow is compiled before activation. If compilation fails, the workflow is +not registered and the compiler diagnostics are returned. diff --git a/internal/harness/tools/descriptions/embed_test.go b/internal/harness/tools/descriptions/embed_test.go index 709e82f..d489608 100644 --- a/internal/harness/tools/descriptions/embed_test.go +++ b/internal/harness/tools/descriptions/embed_test.go @@ -70,6 +70,7 @@ func TestLoadAllKnownDescriptions(t *testing.T) { "create_profile", "create_prompt_extension", "create_skill", + "create_workflow", "cron_create", "cron_delete", "cron_get", @@ -114,6 +115,7 @@ func TestLoadAllKnownDescriptions(t *testing.T) { "reset_context", "run_agent", "run_recipe", + "run_workflow", "search_conversations", "set_delayed_callback", "skill", @@ -185,6 +187,7 @@ func TestEmbeddedFSAndKnownListAreInSync(t *testing.T) { "create_profile": true, "create_prompt_extension": true, "create_skill": true, + "create_workflow": true, "cron_create": true, "cron_delete": true, "cron_get": true, @@ -231,6 +234,7 @@ func TestEmbeddedFSAndKnownListAreInSync(t *testing.T) { "run_agent": true, "start_subagent": true, "run_recipe": true, + "run_workflow": true, "wait_subagent": true, "search_conversations": true, "set_delayed_callback": true, diff --git a/internal/harness/tools/descriptions/run_workflow.md b/internal/harness/tools/descriptions/run_workflow.md new file mode 100644 index 0000000..8735f33 --- /dev/null +++ b/internal/harness/tools/descriptions/run_workflow.md @@ -0,0 +1,15 @@ +Run or resume a registered Go-authored workflow. + +By default this tool waits for the workflow to finish and returns: +- run id +- status +- result JSON +- error, if any +- structured workflow feedback events + +Workflow feedback includes progress, findings, warnings, questions, logs, and +phase changes emitted by the workflow through the SDK. Use this feedback to +reason about intermediate work instead of waiting for only a final answer. + +Set `wait=false` to start the workflow and poll through the workflow run API. +Use `resume_run_id` to retry a failed workflow run with new args. diff --git a/internal/harness/tools_default.go b/internal/harness/tools_default.go index 7ba0197..e23d822 100644 --- a/internal/harness/tools_default.go +++ b/internal/harness/tools_default.go @@ -15,6 +15,7 @@ import ( om "go-agent-harness/internal/observationalmemory" "go-agent-harness/internal/provider/catalog" "go-agent-harness/internal/skills/packs" + "go-agent-harness/internal/workflow" "go-agent-harness/internal/workingmemory" ) @@ -41,6 +42,7 @@ type DefaultRegistryOptions struct { PromptExtensionDirs htools.PromptExtensionDirs // directories for create_prompt_extension tool PackRegistry *packs.PackRegistry // optional skill pack registry ScriptToolsDir string // optional: directory containing user script tools + WorkflowService workflow.SourceService // optional: enables create_workflow and run_workflow ConversationStore ConversationStore // optional: enables list_conversations and search_conversations MessageSummarizer htools.MessageSummarizer // optional: enables summarize/hybrid modes in compact_history // SubagentManager enables the run_agent tool for profile-based subagent delegation. @@ -326,6 +328,13 @@ func NewDefaultRegistryWithOptions(workspaceRoot string, opts DefaultRegistryOpt } } + if opts.WorkflowService != nil { + deferredTools = append(deferredTools, + deferred.CreateWorkflowTool(opts.WorkflowService), + deferred.RunWorkflowTool(opts.WorkflowService), + ) + } + // create_skill tool: available whenever a skills directory is configured. if opts.SkillsDir != "" { deferredTools = append(deferredTools, deferred.CreateSkillTool(opts.SkillsDir)) diff --git a/internal/server/http_script_workflows_test.go b/internal/server/http_script_workflows_test.go index e028374..04ed851 100644 --- a/internal/server/http_script_workflows_test.go +++ b/internal/server/http_script_workflows_test.go @@ -636,6 +636,41 @@ func TestPOC9_SSEEventFormat(t *testing.T) { assert.GreaterOrEqual(t, eventCount, 2, "should have at least started + completed events") } +func TestScriptWorkflowSSEIncludesFeedbackHistory(t *testing.T) { + mgr := newMockScriptWorkflowMgr() + srv := newTestServerWithScriptWorkflows(mgr) + + mux := http.NewServeMux() + srv.registerScriptWorkflowRoutes(mux, testAuth) + + req := httptest.NewRequest(http.MethodPost, "/v1/script-workflows/test-workflow/runs", nil) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + require.Equal(t, http.StatusAccepted, rec.Code) + var startResp struct { + RunID string `json:"run_id"` + } + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &startResp)) + + mgr.emit(startResp.RunID, workflow.EventWorkflowFinding, map[string]any{ + "kind": "finding", + "message": "found evidence", + "data": map[string]any{"file": "main.go"}, + "requires_response": false, + }) + + req = httptest.NewRequest(http.MethodGet, "/v1/script-workflow-runs/"+startResp.RunID+"/events", nil) + rec = httptest.NewRecorder() + mux.ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code) + + body := rec.Body.String() + require.Contains(t, body, "event: workflow.finding") + require.Contains(t, body, `"kind":"finding"`) + require.Contains(t, body, `"message":"found evidence"`) + require.Contains(t, body, `"requires_response":false`) +} + // ============================================================================= // POC 10: Nil manager returns 501 Not Implemented // ============================================================================= diff --git a/internal/workflow/context.go b/internal/workflow/context.go index 4db7ef9..7165182 100644 --- a/internal/workflow/context.go +++ b/internal/workflow/context.go @@ -5,6 +5,8 @@ import ( "fmt" "runtime" "sync" + + "github.com/google/uuid" ) // Context is the execution context passed to workflow scripts. @@ -26,8 +28,8 @@ type Context struct { mu sync.Mutex wg sync.WaitGroup - sem chan struct{} // concurrency semaphore - results []AgentResult // accumulated agent call results + sem chan struct{} // concurrency semaphore + results []AgentResult // accumulated agent call results } // newContext creates a Context for a workflow run. @@ -88,10 +90,16 @@ func (c *Context) Agent(prompt string, opts *AgentOpts) (*AgentResult, error) { }) req := SubagentRequest{ - Prompt: prompt, - Model: opts.Model, - Isolation: opts.Isolation, - AgentType: opts.AgentType, + Prompt: prompt, + Model: opts.Model, + Provider: opts.Provider, + Profile: opts.Profile, + AllowedTools: append([]string(nil), opts.AllowedTools...), + Isolation: opts.Isolation, + CleanupPolicy: opts.CleanupPolicy, + AgentType: opts.AgentType, + MaxSteps: opts.MaxSteps, + MaxCostUSD: opts.MaxCostUSD, } result, err := c.engine.subagents.Create(c.ctx, req) @@ -331,6 +339,54 @@ func (c *Context) Log(message string) { }) } +// Feedback emits structured progress, findings, warnings, or debug messages +// back to the parent agent and API subscribers. +func (c *Context) Feedback(kind, message string, data map[string]any) { + if kind == "" { + kind = "progress" + } + eventType := EventWorkflowFeedback + switch kind { + case "finding": + eventType = EventWorkflowFinding + case "warning": + eventType = EventWorkflowWarning + } + payload := map[string]any{ + "kind": kind, + "message": message, + "requires_response": false, + } + if data != nil { + payload["data"] = data + } + c.emit(eventType, payload) +} + +// Question emits a structured question and asks the engine's configured +// responder for an answer. Without a responder the question is visible to +// subscribers and the workflow receives a clear error. +func (c *Context) Question(prompt string, choices []QuestionOption) (any, error) { + callID := "workflow_question_" + uuid.NewString() + payload := map[string]any{ + "kind": "question", + "message": prompt, + "requires_response": true, + "call_id": callID, + "choices": choices, + } + c.emit(EventWorkflowQuestion, payload) + if c.engine.questions == nil { + return nil, fmt.Errorf("workflow question responder is not configured") + } + return c.engine.questions.AskWorkflowQuestion(c.ctx, QuestionRequest{ + RunID: c.runID, + CallID: callID, + Prompt: prompt, + Choices: choices, + }) +} + // Workflow runs a nested workflow by name. The nested workflow shares this // context's budget (via Clone) and concurrency pool. // diff --git a/internal/workflow/engine.go b/internal/workflow/engine.go index 64832f8..efee270 100644 --- a/internal/workflow/engine.go +++ b/internal/workflow/engine.go @@ -29,6 +29,11 @@ type EngineOptions struct { // is used. The store persists workflow runs and events. Store Store + // QuestionResponder handles workflow questions that need a parent/user + // answer. When nil, Context.Question returns an error after emitting the + // question event. + QuestionResponder QuestionResponder + // Now overrides the time source for deterministic testing. Now func() time.Time } @@ -45,6 +50,7 @@ type Engine struct { maxConcurrency int defaultBudget int store Store + questions QuestionResponder now func() time.Time mu sync.Mutex @@ -79,6 +85,7 @@ func NewEngine(opts EngineOptions) *Engine { maxConcurrency: concurrency, defaultBudget: opts.DefaultBudget, store: opts.Store, + questions: opts.QuestionResponder, now: opts.Now, subs: make(map[string]map[chan Event]struct{}), eventSeqs: make(map[string]int64), @@ -93,10 +100,18 @@ func NewEngine(opts EngineOptions) *Engine { // The name is used as the workflow's Meta.Name for event emission, // progress display, and nested workflow identification. func (e *Engine) Register(name string, script Script) { + e.RegisterWithMeta(Meta{Name: name}, script) +} + +// RegisterWithMeta adds a workflow script with explicit discovery metadata. +func (e *Engine) RegisterWithMeta(meta Meta, script Script) { + if meta.Name == "" { + return + } e.mu.Lock() defer e.mu.Unlock() - e.scripts[name] = registeredScript{ - Meta: Meta{Name: name}, + e.scripts[meta.Name] = registeredScript{ + Meta: meta, Script: script, } } diff --git a/internal/workflow/source.go b/internal/workflow/source.go new file mode 100644 index 0000000..5c12efe --- /dev/null +++ b/internal/workflow/source.go @@ -0,0 +1,832 @@ +package workflow + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + "os" + "os/exec" + "path/filepath" + "regexp" + "strings" + "sync" + "syscall" + "time" +) + +const ( + sourceManifestName = "workflow.json" + defaultWorkflowTimeout = 5 * time.Minute + maxWorkflowProtocolBytes = 1024 * 1024 + maxWorkflowStderrBytes = 32 * 1024 +) + +var sourceWorkflowNameRe = regexp.MustCompile(`^[a-z0-9]+(-[a-z0-9]+)*$`) + +// SourceManager discovers, builds, registers, and runs Go-authored workflow +// bundles without requiring the harness process to restart. +type SourceManager struct { + engine *Engine + + workflowDirs []string + skillDirs []string + cacheDir string + moduleRoot string + goBinary string + + mu sync.RWMutex + bundles map[string]*SourceBundle +} + +type SourceService interface { + List() []Meta + Start(ctx context.Context, name string, args any) (*Run, error) + Resume(ctx context.Context, runID string, args any) (*Run, error) + GetRun(runID string) (*Run, error) + Subscribe(runID string) ([]Event, <-chan Event, func(), error) + Wait(ctx context.Context, runID string) (*Run, []Event, error) + CreateWorkflow(ctx context.Context, req CreateWorkflowRequest) (*SourceBundle, error) +} + +type SourceManagerOptions struct { + Engine *Engine + WorkflowDirs []string + SkillDirs []string + CacheDir string + ModuleRoot string + GoBinary string +} + +type SourceBundleManifest struct { + Name string `json:"name"` + Description string `json:"description"` + Version int `json:"version"` + Language string `json:"language"` + Entrypoint string `json:"entrypoint"` + WhenToUse string `json:"when_to_use,omitempty"` + ArgsSchema map[string]any `json:"args_schema,omitempty"` + Skill string `json:"skill,omitempty"` + TimeoutSeconds int `json:"timeout_seconds,omitempty"` +} + +type SourceBundle struct { + Manifest SourceBundleManifest `json:"manifest"` + Dir string `json:"dir"` + Hash string `json:"hash"` + Binary string `json:"binary"` + Scope string `json:"scope"` +} + +type CreateWorkflowRequest struct { + Name string `json:"name"` + Description string `json:"description"` + WhenToUse string `json:"when_to_use,omitempty"` + Source string `json:"source"` + ArgsSchema map[string]any `json:"args_schema,omitempty"` + Scope string `json:"scope,omitempty"` + Skill string `json:"skill,omitempty"` + TimeoutSeconds int `json:"timeout_seconds,omitempty"` + Overwrite bool `json:"overwrite,omitempty"` +} + +func NewSourceManager(opts SourceManagerOptions) (*SourceManager, error) { + if opts.Engine == nil { + return nil, fmt.Errorf("workflow source manager requires an engine") + } + if opts.GoBinary == "" { + opts.GoBinary = "go" + } + if opts.CacheDir == "" { + opts.CacheDir = filepath.Join(os.TempDir(), "go-agent-harness-workflows") + } + moduleRoot := strings.TrimSpace(opts.ModuleRoot) + if moduleRoot == "" { + moduleRoot = findModuleRoot() + } + return &SourceManager{ + engine: opts.Engine, + workflowDirs: cleanDirs(opts.WorkflowDirs), + skillDirs: cleanDirs(opts.SkillDirs), + cacheDir: opts.CacheDir, + moduleRoot: moduleRoot, + goBinary: opts.GoBinary, + bundles: make(map[string]*SourceBundle), + }, nil +} + +func cleanDirs(dirs []string) []string { + out := make([]string, 0, len(dirs)) + for _, dir := range dirs { + if strings.TrimSpace(dir) != "" { + out = append(out, dir) + } + } + return out +} + +func (m *SourceManager) Load(ctx context.Context) error { + bundles, err := m.discover(ctx) + if err != nil { + return err + } + for _, bundle := range bundles { + if err := m.build(ctx, bundle); err != nil { + continue + } + m.register(bundle) + } + return nil +} + +func (m *SourceManager) List() []Meta { + return m.engine.List() +} + +func (m *SourceManager) Start(ctx context.Context, name string, args any) (*Run, error) { + return m.engine.Start(ctx, name, args) +} + +func (m *SourceManager) Resume(ctx context.Context, runID string, args any) (*Run, error) { + return m.engine.Resume(ctx, runID, args) +} + +func (m *SourceManager) GetRun(runID string) (*Run, error) { + return m.engine.GetRun(runID) +} + +func (m *SourceManager) Subscribe(runID string) ([]Event, <-chan Event, func(), error) { + return m.engine.Subscribe(runID) +} + +func (m *SourceManager) Wait(ctx context.Context, runID string) (*Run, []Event, error) { + history, stream, cancel, err := m.Subscribe(runID) + if err != nil { + return nil, nil, err + } + defer cancel() + events := append([]Event(nil), history...) + if run, err := m.GetRun(runID); err == nil && isTerminalRun(run.Status) { + return run, events, nil + } + for { + select { + case <-ctx.Done(): + run, _ := m.GetRun(runID) + return run, events, ctx.Err() + case ev, ok := <-stream: + if !ok { + run, _ := m.GetRun(runID) + return run, events, nil + } + events = append(events, ev) + if ev.Type == EventWorkflowCompleted || ev.Type == EventWorkflowFailed { + run, _ := m.GetRun(runID) + return run, events, nil + } + } + } +} + +func isTerminalRun(status RunStatus) bool { + return status == RunStatusCompleted || status == RunStatusFailed +} + +func (m *SourceManager) CreateWorkflow(ctx context.Context, req CreateWorkflowRequest) (*SourceBundle, error) { + name := strings.TrimSpace(req.Name) + if !sourceWorkflowNameRe.MatchString(name) { + return nil, fmt.Errorf("workflow name %q must be kebab-case", req.Name) + } + if strings.TrimSpace(req.Description) == "" { + return nil, fmt.Errorf("description is required") + } + if strings.TrimSpace(req.Source) == "" { + return nil, fmt.Errorf("source is required") + } + dir, scope, err := m.createDir(req) + if err != nil { + return nil, err + } + if _, statErr := os.Stat(dir); statErr == nil && !req.Overwrite { + return nil, fmt.Errorf("workflow %q already exists at %s", name, dir) + } + if req.Overwrite { + if err := os.RemoveAll(dir); err != nil { + return nil, err + } + } + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + manifest := SourceBundleManifest{ + Name: name, + Description: strings.TrimSpace(req.Description), + Version: 1, + Language: "go", + Entrypoint: "main.go", + WhenToUse: strings.TrimSpace(req.WhenToUse), + ArgsSchema: req.ArgsSchema, + Skill: strings.TrimSpace(req.Skill), + TimeoutSeconds: req.TimeoutSeconds, + } + raw, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + return nil, err + } + if err := os.WriteFile(filepath.Join(dir, sourceManifestName), append(raw, '\n'), 0o644); err != nil { + return nil, err + } + if err := os.WriteFile(filepath.Join(dir, manifest.Entrypoint), []byte(strings.TrimSpace(req.Source)+"\n"), 0o644); err != nil { + return nil, err + } + bundle, err := m.loadBundle(dir, scope) + if err != nil { + return nil, err + } + if err := m.build(ctx, bundle); err != nil { + return nil, err + } + m.register(bundle) + return bundle, nil +} + +func (m *SourceManager) createDir(req CreateWorkflowRequest) (string, string, error) { + scope := strings.TrimSpace(req.Scope) + if scope == "" { + scope = "workspace" + } + switch scope { + case "workspace": + if len(m.workflowDirs) == 0 { + return "", "", fmt.Errorf("workspace workflow directory is not configured") + } + return filepath.Join(m.workflowDirs[len(m.workflowDirs)-1], req.Name), scope, nil + case "global": + if len(m.workflowDirs) == 0 { + return "", "", fmt.Errorf("global workflow directory is not configured") + } + return filepath.Join(m.workflowDirs[0], req.Name), scope, nil + case "skill": + skill := strings.TrimSpace(req.Skill) + if skill == "" { + return "", "", fmt.Errorf("skill is required when scope=skill") + } + for i := len(m.skillDirs) - 1; i >= 0; i-- { + root := m.skillDirs[i] + if root == "" { + continue + } + return filepath.Join(root, skill, "workflows", req.Name), scope, nil + } + return "", "", fmt.Errorf("skill workflow directory is not configured") + default: + return "", "", fmt.Errorf("unsupported workflow scope %q", scope) + } +} + +func (m *SourceManager) discover(ctx context.Context) ([]*SourceBundle, error) { + var out []*SourceBundle + for _, root := range m.workflowDirs { + if err := ctx.Err(); err != nil { + return nil, err + } + bundles, err := discoverDirectBundles(root, "workflow") + if err != nil { + return nil, err + } + out = append(out, bundles...) + } + for _, skillRoot := range m.skillDirs { + if err := ctx.Err(); err != nil { + return nil, err + } + bundles, err := discoverSkillBundles(skillRoot) + if err != nil { + return nil, err + } + out = append(out, bundles...) + } + return out, nil +} + +func discoverDirectBundles(root, scope string) ([]*SourceBundle, error) { + entries, err := os.ReadDir(root) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return nil, err + } + var out []*SourceBundle + for _, entry := range entries { + if !entry.IsDir() { + continue + } + bundle, err := loadSourceBundle(filepath.Join(root, entry.Name()), scope) + if err != nil { + continue + } + if bundle != nil { + out = append(out, bundle) + } + } + return out, nil +} + +func discoverSkillBundles(skillRoot string) ([]*SourceBundle, error) { + entries, err := os.ReadDir(skillRoot) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return nil, err + } + var out []*SourceBundle + for _, skill := range entries { + if !skill.IsDir() { + continue + } + root := filepath.Join(skillRoot, skill.Name(), "workflows") + bundles, err := discoverDirectBundles(root, "skill") + if err != nil { + return nil, err + } + for _, bundle := range bundles { + if bundle.Manifest.Skill == "" { + bundle.Manifest.Skill = skill.Name() + } + out = append(out, bundle) + } + } + return out, nil +} + +func (m *SourceManager) loadBundle(dir, scope string) (*SourceBundle, error) { + return loadSourceBundle(dir, scope) +} + +func loadSourceBundle(dir, scope string) (*SourceBundle, error) { + manifestPath := filepath.Join(dir, sourceManifestName) + raw, err := os.ReadFile(manifestPath) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return nil, err + } + var manifest SourceBundleManifest + if err := json.Unmarshal(raw, &manifest); err != nil { + return nil, fmt.Errorf("parse %s: %w", manifestPath, err) + } + if err := validateSourceManifest(manifest); err != nil { + return nil, fmt.Errorf("%s: %w", manifestPath, err) + } + return &SourceBundle{Manifest: manifest, Dir: dir, Scope: scope}, nil +} + +func validateSourceManifest(manifest SourceBundleManifest) error { + if !sourceWorkflowNameRe.MatchString(manifest.Name) { + return fmt.Errorf("name %q must be kebab-case", manifest.Name) + } + if strings.TrimSpace(manifest.Description) == "" { + return fmt.Errorf("description is required") + } + if manifest.Version != 1 { + return fmt.Errorf("version must be 1") + } + if manifest.Language != "go" { + return fmt.Errorf("language must be go") + } + if strings.TrimSpace(manifest.Entrypoint) == "" { + return fmt.Errorf("entrypoint is required") + } + return nil +} + +func (m *SourceManager) register(bundle *SourceBundle) { + b := *bundle + m.mu.Lock() + m.bundles[b.Manifest.Name] = &b + m.mu.Unlock() + m.engine.RegisterWithMeta(Meta{ + Name: b.Manifest.Name, + Description: b.Manifest.Description, + WhenToUse: b.Manifest.WhenToUse, + }, func(ctx *Context) (any, error) { + return m.runSourceWorkflow(ctx, &b) + }) +} + +func (m *SourceManager) build(ctx context.Context, bundle *SourceBundle) error { + hash, err := hashBundle(bundle.Dir) + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Join(m.cacheDir, "bin"), 0o755); err != nil { + return err + } + binary := filepath.Join(m.cacheDir, "bin", bundle.Manifest.Name+"-"+hash) + bundle.Hash = hash + bundle.Binary = binary + if _, err := os.Stat(binary); err == nil { + return nil + } + if strings.TrimSpace(m.moduleRoot) == "" { + return fmt.Errorf("harness module root is not configured; set HARNESS_SOURCE_ROOT") + } + buildDir := filepath.Join(m.cacheDir, "src", bundle.Manifest.Name+"-"+hash) + if err := os.RemoveAll(buildDir); err != nil { + return err + } + if err := copyTree(bundle.Dir, buildDir); err != nil { + return err + } + mod := fmt.Sprintf("module workflow.local/%s\n\ngo 1.25.0\n\nrequire go-agent-harness v0.0.0\n\nreplace go-agent-harness => %s\n", + bundle.Manifest.Name, filepath.ToSlash(m.moduleRoot)) + if err := os.WriteFile(filepath.Join(buildDir, "go.mod"), []byte(mod), 0o644); err != nil { + return err + } + cmd := exec.CommandContext(ctx, m.goBinary, "build", "-o", binary, ".") + cmd.Dir = buildDir + cmd.Env = minimalGoEnv() + var stderr bytes.Buffer + cmd.Stderr = &stderr + if err := cmd.Run(); err != nil { + return fmt.Errorf("build workflow %q: %w: %s", bundle.Manifest.Name, err, boundedString(stderr.String(), maxWorkflowStderrBytes)) + } + return nil +} + +func (m *SourceManager) runSourceWorkflow(ctx *Context, bundle *SourceBundle) (any, error) { + if err := m.build(ctx.ctx, bundle); err != nil { + return nil, err + } + timeout := defaultWorkflowTimeout + if bundle.Manifest.TimeoutSeconds > 0 { + timeout = time.Duration(bundle.Manifest.TimeoutSeconds) * time.Second + } + runCtx, cancel := context.WithTimeout(ctx.ctx, timeout) + defer cancel() + + cmd := exec.CommandContext(runCtx, bundle.Binary) + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.Env = minimalChildEnv() + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, err + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, err + } + var stderr bytes.Buffer + cmd.Stderr = &limitedWriter{w: &stderr, max: maxWorkflowStderrBytes} + if err := cmd.Start(); err != nil { + return nil, err + } + enc := json.NewEncoder(stdin) + if err := enc.Encode(protocolResponse{Type: "start", Result: mustRaw(ctx.Args)}); err != nil { + _ = killProcessGroup(cmd) + return nil, err + } + + result, protocolErr := m.serveProtocol(runCtx, ctx, stdout, enc) + if protocolErr != nil { + _ = killProcessGroup(cmd) + } + closeErr := stdin.Close() + waitErr := cmd.Wait() + if runCtx.Err() == context.DeadlineExceeded { + _ = killProcessGroup(cmd) + return nil, fmt.Errorf("workflow %q timed out after %s", bundle.Manifest.Name, timeout) + } + if protocolErr != nil { + _ = killProcessGroup(cmd) + return nil, protocolErr + } + if closeErr != nil { + return nil, closeErr + } + if waitErr != nil { + return nil, fmt.Errorf("workflow %q exited: %w: %s", bundle.Manifest.Name, waitErr, boundedString(stderr.String(), maxWorkflowStderrBytes)) + } + if result == nil { + return nil, fmt.Errorf("workflow %q exited without a result", bundle.Manifest.Name) + } + return result, nil +} + +type protocolMessage struct { + ID string `json:"id,omitempty"` + Type string `json:"type"` + Args json.RawMessage `json:"args,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +type protocolResponse struct { + ID string `json:"id,omitempty"` + Type string `json:"type,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +func (m *SourceManager) serveProtocol(runCtx context.Context, ctx *Context, stdout io.Reader, enc *json.Encoder) (any, error) { + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 0, 64*1024), maxWorkflowProtocolBytes) + var result any + terminal := false + for scanner.Scan() { + var msg protocolMessage + line := scanner.Bytes() + if err := json.Unmarshal(line, &msg); err != nil { + return nil, fmt.Errorf("workflow protocol error: invalid json: %w", err) + } + if terminal { + return nil, fmt.Errorf("workflow protocol error: message after terminal result") + } + switch msg.Type { + case "result": + if result != nil { + return nil, fmt.Errorf("workflow protocol error: duplicate result") + } + terminal = true + if len(msg.Result) > 0 { + if err := json.Unmarshal(msg.Result, &result); err != nil { + return nil, err + } + } else { + result = map[string]any{} + } + case "error": + if msg.Error == "" { + msg.Error = "workflow child returned an error" + } + return nil, fmt.Errorf("%s", msg.Error) + case "phase": + var args struct { + Title string `json:"title"` + } + if err := decodeArgs(msg.Args, &args); err != nil { + return nil, err + } + ctx.Phase(args.Title) + if err := enc.Encode(protocolResponse{ID: msg.ID, Result: mustRaw(true)}); err != nil { + return nil, err + } + case "log": + var args struct { + Message string `json:"message"` + } + if err := decodeArgs(msg.Args, &args); err != nil { + return nil, err + } + ctx.Log(args.Message) + if err := enc.Encode(protocolResponse{ID: msg.ID, Result: mustRaw(true)}); err != nil { + return nil, err + } + case "feedback": + var args struct { + Kind string `json:"kind"` + Message string `json:"message"` + Data map[string]any `json:"data"` + } + if err := decodeArgs(msg.Args, &args); err != nil { + return nil, err + } + ctx.Feedback(args.Kind, args.Message, args.Data) + if err := enc.Encode(protocolResponse{ID: msg.ID, Result: mustRaw(true)}); err != nil { + return nil, err + } + case "agent": + resp, err := m.handleAgent(runCtx, ctx, msg.Args) + if err := writeProtocolResponse(enc, msg.ID, resp, err); err != nil { + return nil, err + } + case "workflow": + resp, err := m.handleNestedWorkflow(ctx, msg.Args) + if err := writeProtocolResponse(enc, msg.ID, resp, err); err != nil { + return nil, err + } + case "question": + resp, err := m.handleQuestion(ctx, msg.Args) + if err := writeProtocolResponse(enc, msg.ID, resp, err); err != nil { + return nil, err + } + default: + return nil, fmt.Errorf("workflow protocol error: unknown message type %q", msg.Type) + } + } + if err := scanner.Err(); err != nil { + return nil, err + } + return result, nil +} + +func (m *SourceManager) handleAgent(_ context.Context, ctx *Context, raw json.RawMessage) (*AgentResult, error) { + var args struct { + Prompt string `json:"prompt"` + Opts *AgentOpts `json:"opts"` + } + if err := decodeArgs(raw, &args); err != nil { + return nil, err + } + return ctx.Agent(args.Prompt, args.Opts) +} + +func (m *SourceManager) handleNestedWorkflow(ctx *Context, raw json.RawMessage) (any, error) { + var args struct { + Name string `json:"name"` + Args any `json:"args"` + } + if err := decodeArgs(raw, &args); err != nil { + return nil, err + } + return ctx.Workflow(args.Name, args.Args) +} + +func (m *SourceManager) handleQuestion(ctx *Context, raw json.RawMessage) (any, error) { + var args struct { + Prompt string `json:"prompt"` + Choices []QuestionOption `json:"choices"` + } + if err := decodeArgs(raw, &args); err != nil { + return nil, err + } + return ctx.Question(args.Prompt, args.Choices) +} + +func decodeArgs(raw json.RawMessage, out any) error { + if len(raw) == 0 { + return nil + } + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("decode workflow protocol args: %w", err) + } + return nil +} + +func writeProtocolResponse(enc *json.Encoder, id string, result any, err error) error { + resp := protocolResponse{ID: id} + if err != nil { + resp.Error = err.Error() + } else { + resp.Result = mustRaw(result) + } + return enc.Encode(resp) +} + +func mustRaw(value any) json.RawMessage { + raw, err := json.Marshal(value) + if err != nil { + return json.RawMessage(`null`) + } + return raw +} + +func hashBundle(dir string) (string, error) { + h := sha256.New() + err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + return nil + } + rel, err := filepath.Rel(dir, path) + if err != nil { + return err + } + h.Write([]byte(rel)) + raw, err := os.ReadFile(path) + if err != nil { + return err + } + h.Write(raw) + return nil + }) + if err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil))[:16], nil +} + +func copyTree(src, dst string) error { + return filepath.WalkDir(src, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + rel, err := filepath.Rel(src, path) + if err != nil { + return err + } + target := filepath.Join(dst, rel) + if d.IsDir() { + return os.MkdirAll(target, 0o755) + } + raw, err := os.ReadFile(path) + if err != nil { + return err + } + info, err := d.Info() + if err != nil { + return err + } + return os.WriteFile(target, raw, info.Mode()) + }) +} + +func minimalGoEnv() []string { + env := minimalChildEnv() + env = append(env, "GOWORK=off") + return env +} + +func minimalChildEnv() []string { + return []string{ + "HOME=" + os.Getenv("HOME"), + "PATH=" + os.Getenv("PATH"), + } +} + +func killProcessGroup(cmd *exec.Cmd) error { + if cmd == nil || cmd.Process == nil { + return nil + } + return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) +} + +func boundedString(s string, max int) string { + if max <= 0 || len(s) <= max { + return strings.TrimSpace(s) + } + return strings.TrimSpace(s[:max]) + "...[truncated]" +} + +type limitedWriter struct { + w io.Writer + max int + n int +} + +func (w *limitedWriter) Write(p []byte) (int, error) { + if w.max <= 0 || w.n >= w.max { + return len(p), nil + } + remaining := w.max - w.n + write := p + if len(write) > remaining { + write = write[:remaining] + } + n, err := w.w.Write(write) + w.n += n + if err != nil { + return n, err + } + return len(p), nil +} + +func findModuleRoot() string { + candidates := []string{os.Getenv("HARNESS_SOURCE_ROOT")} + if cwd, err := os.Getwd(); err == nil { + candidates = append(candidates, cwd) + } + for _, start := range candidates { + if root := findModuleRootFrom(start); root != "" { + return root + } + } + return "" +} + +func findModuleRootFrom(start string) string { + if strings.TrimSpace(start) == "" { + return "" + } + current, err := filepath.Abs(start) + if err != nil { + return "" + } + info, err := os.Stat(current) + if err == nil && !info.IsDir() { + current = filepath.Dir(current) + } + for { + mod := filepath.Join(current, "go.mod") + raw, err := os.ReadFile(mod) + if err == nil && strings.Contains(string(raw), "module go-agent-harness") { + return current + } + parent := filepath.Dir(current) + if parent == current { + return "" + } + current = parent + } +} diff --git a/internal/workflow/source_test.go b/internal/workflow/source_test.go new file mode 100644 index 0000000..edd84ab --- /dev/null +++ b/internal/workflow/source_test.go @@ -0,0 +1,528 @@ +package workflow_test + +import ( + "context" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "go-agent-harness/internal/workflow" +) + +type sourceNoopSubagents struct{} + +func (sourceNoopSubagents) Create(context.Context, workflow.SubagentRequest) (workflow.SubagentResult, error) { + return workflow.SubagentResult{ID: "subagent_1", Status: "completed", Output: "ok"}, nil +} + +func (sourceNoopSubagents) Get(context.Context, string) (workflow.SubagentResult, error) { + return workflow.SubagentResult{ID: "subagent_1", Status: "completed", Output: "ok"}, nil +} + +type sourceRecordingSubagents struct { + mu sync.Mutex + req workflow.SubagentRequest +} + +func (s *sourceRecordingSubagents) Create(_ context.Context, req workflow.SubagentRequest) (workflow.SubagentResult, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.req = req + return workflow.SubagentResult{ID: "subagent_1", Status: "completed", Output: "agent-output"}, nil +} + +func (s *sourceRecordingSubagents) Get(context.Context, string) (workflow.SubagentResult, error) { + return workflow.SubagentResult{ID: "subagent_1", Status: "completed", Output: "agent-output"}, nil +} + +func (s *sourceRecordingSubagents) request() workflow.SubagentRequest { + s.mu.Lock() + defer s.mu.Unlock() + return s.req +} + +type sourceQuestionResponder struct { + mu sync.Mutex + req workflow.QuestionRequest +} + +func (r *sourceQuestionResponder) AskWorkflowQuestion(_ context.Context, req workflow.QuestionRequest) (any, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.req = req + return "Continue", nil +} + +func (r *sourceQuestionResponder) request() workflow.QuestionRequest { + r.mu.Lock() + defer r.mu.Unlock() + return r.req +} + +func TestSourceManagerCreateWorkflowHotRegistersAndRunsFeedback(t *testing.T) { + root := t.TempDir() + engine := workflow.NewEngine(workflow.EngineOptions{Subagents: sourceNoopSubagents{}}) + manager, err := workflow.NewSourceManager(workflow.SourceManagerOptions{ + Engine: engine, + WorkflowDirs: []string{filepath.Join(root, "global"), filepath.Join(root, "workspace")}, + CacheDir: filepath.Join(root, "cache"), + ModuleRoot: mustRepoRoot(t), + }) + require.NoError(t, err) + + source := `package main + +import sdk "go-agent-harness/pkg/workflowsdk" + +func main() { + sdk.Main(func(ctx *sdk.Context) (any, error) { + _ = ctx.Phase("Inspect") + _ = ctx.Feedback("finding", "found useful evidence", map[string]any{"file": "main.go"}) + return map[string]any{"ok": true, "args": ctx.Args}, nil + }) +}` + + bundle, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "evidence-check", + Description: "Checks evidence.", + WhenToUse: "Use when evidence needs validation.", + Source: source, + Scope: "workspace", + }) + require.NoError(t, err) + require.Equal(t, "evidence-check", bundle.Manifest.Name) + require.NotEmpty(t, bundle.Hash) + + run, err := manager.Start(context.Background(), "evidence-check", map[string]any{"target": "x"}) + require.NoError(t, err) + + waitCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + final, events, err := manager.Wait(waitCtx, run.ID) + require.NoError(t, err) + require.Equal(t, workflow.RunStatusCompleted, final.Status) + require.Contains(t, final.ResultJSON, `"ok":true`) + + var sawFinding bool + for _, ev := range events { + if ev.Type == workflow.EventWorkflowFinding { + sawFinding = true + require.Equal(t, "finding", ev.Payload["kind"]) + require.Equal(t, "found useful evidence", ev.Payload["message"]) + } + } + require.True(t, sawFinding, "expected workflow finding feedback event") +} + +func TestSourceManagerCreateWorkflowRejectsDuplicateUntilOverwrite(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "daily-review", + Description: "Review the day.", + Source: workflowSourceReturning(`map[string]any{"version": 1}`), + Scope: "workspace", + }) + require.NoError(t, err) + + _, err = manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "daily-review", + Description: "Review the day again.", + Source: workflowSourceReturning(`map[string]any{"version": 2}`), + Scope: "workspace", + }) + require.ErrorContains(t, err, "already exists") + + _, err = manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "daily-review", + Description: "Review the day again.", + Source: workflowSourceReturning(`map[string]any{"version": 2}`), + Scope: "workspace", + Overwrite: true, + }) + require.NoError(t, err) + + final := runWorkflowToTerminal(t, manager, "daily-review", map[string]any{}) + require.Equal(t, workflow.RunStatusCompleted, final.Status) + require.Contains(t, final.ResultJSON, `"version":2`) +} + +func TestSourceManagerCreateWorkflowCompileFailureDoesNotActivate(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "broken-workflow", + Description: "Does not compile.", + Source: "package main\nfunc main() {", + Scope: "workspace", + }) + require.ErrorContains(t, err, "build workflow") + + _, err = manager.Start(context.Background(), "broken-workflow", map[string]any{}) + require.ErrorContains(t, err, "not found") +} + +func TestSourceManagerCreateWorkflowValidatesName(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "Not Kebab", + Description: "Invalid name.", + Source: workflowSourceReturning(`"ok"`), + Scope: "workspace", + }) + require.ErrorContains(t, err, "must be kebab-case") +} + +func TestSourceManagerRunWorkflowFailsOnInvalidProtocolAfterResult(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + source := `package main + +import "fmt" + +func main() { + fmt.Println(` + "`" + `{"type":"result","result":{"ok":true}}` + "`" + `) + fmt.Println(` + "`" + `{"type":"log","args":{"message":"late side effect"}}` + "`" + `) +}` + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "bad-protocol", + Description: "Emits a protocol message after result.", + Source: source, + Scope: "workspace", + }) + require.NoError(t, err) + + final := runWorkflowToTerminal(t, manager, "bad-protocol", map[string]any{}) + require.Equal(t, workflow.RunStatusFailed, final.Status) + require.Contains(t, final.Error, "message after terminal result") +} + +func TestSourceManagerRunWorkflowFailsOnTimeout(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + source := `package main + +import "time" + +func main() { + time.Sleep(3 * time.Second) +}` + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "sleepy-workflow", + Description: "Sleeps too long.", + Source: source, + Scope: "workspace", + TimeoutSeconds: 1, + }) + require.NoError(t, err) + + final := runWorkflowToTerminal(t, manager, "sleepy-workflow", map[string]any{}) + require.Equal(t, workflow.RunStatusFailed, final.Status) + require.Contains(t, final.Error, "timed out") +} + +func TestSourceManagerRunWorkflowFailsOnProcessExit(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + source := `package main + +import "os" + +func main() { + os.Exit(7) +}` + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "exit-workflow", + Description: "Exits nonzero.", + Source: source, + Scope: "workspace", + }) + require.NoError(t, err) + + final := runWorkflowToTerminal(t, manager, "exit-workflow", map[string]any{}) + require.Equal(t, workflow.RunStatusFailed, final.Status) + require.Contains(t, final.Error, "exited") + + resumed, err := manager.Resume(context.Background(), final.ID, map[string]any{"retry": true}) + require.NoError(t, err) + waitCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + resumedFinal, _, err := manager.Wait(waitCtx, resumed.ID) + require.NoError(t, err) + require.Equal(t, workflow.RunStatusFailed, resumedFinal.Status) +} + +func TestSourceManagerRunWorkflowIncludesBoundedStderrOnProcessExit(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + source := `package main + +import ( + "fmt" + "os" +) + +func main() { + fmt.Fprint(os.Stderr, "child stderr diagnostic") + os.Exit(7) +}` + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "stderr-workflow", + Description: "Writes stderr and exits nonzero.", + Source: source, + Scope: "workspace", + }) + require.NoError(t, err) + + final := runWorkflowToTerminal(t, manager, "stderr-workflow", map[string]any{}) + require.Equal(t, workflow.RunStatusFailed, final.Status) + require.Contains(t, final.Error, "child stderr diagnostic") +} + +func TestSourceManagerNestedWorkflowRPC(t *testing.T) { + root := t.TempDir() + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, nil) + + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "child-workflow", + Description: "Returns a child result.", + Source: workflowSourceReturning(`map[string]any{"child": true}`), + Scope: "workspace", + }) + require.NoError(t, err) + + parent := `package main + +import sdk "go-agent-harness/pkg/workflowsdk" + +func main() { + sdk.Main(func(ctx *sdk.Context) (any, error) { + child, err := ctx.Workflow("child-workflow", map[string]any{"from": "parent"}) + if err != nil { + return nil, err + } + return map[string]any{"child": child}, nil + }) +}` + _, err = manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "parent-workflow", + Description: "Calls a nested workflow.", + Source: parent, + Scope: "workspace", + }) + require.NoError(t, err) + + final := runWorkflowToTerminal(t, manager, "parent-workflow", map[string]any{}) + require.Equal(t, workflow.RunStatusCompleted, final.Status) + require.Contains(t, final.ResultJSON, `"child":true`) +} + +func TestSourceManagerAgentRPCForwardsOptions(t *testing.T) { + root := t.TempDir() + subagents := &sourceRecordingSubagents{} + manager := newTestSourceManager(t, root, subagents, nil) + + source := `package main + +import sdk "go-agent-harness/pkg/workflowsdk" + +func main() { + sdk.Main(func(ctx *sdk.Context) (any, error) { + res, err := ctx.Agent("inspect repository", &sdk.AgentOpts{ + Model: "gpt-5-nano", + Provider: "openai", + Profile: "reviewer", + AllowedTools: []string{"read", "grep"}, + Isolation: "worktree", + CleanupPolicy: "keep", + AgentType: "review", + MaxSteps: 3, + MaxCostUSD: 0.25, + }) + if err != nil { + return nil, err + } + return map[string]any{"output": res.Output}, nil + }) +}` + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "agent-options", + Description: "Calls a subagent with options.", + Source: source, + Scope: "workspace", + }) + require.NoError(t, err) + + final := runWorkflowToTerminal(t, manager, "agent-options", map[string]any{}) + require.Equal(t, workflow.RunStatusCompleted, final.Status) + req := subagents.request() + require.Equal(t, "inspect repository", req.Prompt) + require.Equal(t, "gpt-5-nano", req.Model) + require.Equal(t, "openai", req.Provider) + require.Equal(t, "reviewer", req.Profile) + require.Equal(t, []string{"read", "grep"}, req.AllowedTools) + require.Equal(t, "worktree", req.Isolation) + require.Equal(t, "keep", req.CleanupPolicy) + require.Equal(t, "review", req.AgentType) + require.Equal(t, 3, req.MaxSteps) + require.Equal(t, 0.25, req.MaxCostUSD) +} + +func TestSourceManagerQuestionEmitsEventAndUsesResponder(t *testing.T) { + root := t.TempDir() + responder := &sourceQuestionResponder{} + manager := newTestSourceManager(t, root, sourceNoopSubagents{}, responder) + + source := `package main + +import sdk "go-agent-harness/pkg/workflowsdk" + +func main() { + sdk.Main(func(ctx *sdk.Context) (any, error) { + answer, err := ctx.Question("Continue?", []sdk.QuestionOption{{Label: "Continue", Description: "Keep going."}}) + if err != nil { + return nil, err + } + return map[string]any{"answer": answer}, nil + }) +}` + _, err := manager.CreateWorkflow(context.Background(), workflow.CreateWorkflowRequest{ + Name: "question-workflow", + Description: "Asks a question.", + Source: source, + Scope: "workspace", + }) + require.NoError(t, err) + + run, err := manager.Start(context.Background(), "question-workflow", map[string]any{}) + require.NoError(t, err) + waitCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + final, events, err := manager.Wait(waitCtx, run.ID) + require.NoError(t, err) + require.Equal(t, workflow.RunStatusCompleted, final.Status) + require.Contains(t, final.ResultJSON, `"answer":"Continue"`) + + req := responder.request() + require.Equal(t, "Continue?", req.Prompt) + require.Len(t, req.Choices, 1) + + var sawQuestion bool + for _, ev := range events { + if ev.Type == workflow.EventWorkflowQuestion { + sawQuestion = true + require.Equal(t, "question", ev.Payload["kind"]) + require.Equal(t, true, ev.Payload["requires_response"]) + } + } + require.True(t, sawQuestion, "expected workflow question event") +} + +func TestSourceManagerLoadDiscoversSkillBundledWorkflows(t *testing.T) { + root := t.TempDir() + skillWorkflowDir := filepath.Join(root, "skills", "review", "workflows", "skill-review") + require.NoError(t, os.MkdirAll(skillWorkflowDir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(skillWorkflowDir, "workflow.json"), []byte(`{ + "name": "skill-review", + "description": "Skill bundled review workflow.", + "version": 1, + "language": "go", + "entrypoint": "main.go", + "when_to_use": "Use from the review skill." +}`), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(skillWorkflowDir, "main.go"), []byte(`package main + +import sdk "go-agent-harness/pkg/workflowsdk" + +func main() { + sdk.Main(func(ctx *sdk.Context) (any, error) { + _ = ctx.Log("skill workflow running") + return "done", nil + }) +}`), 0o644)) + + engine := workflow.NewEngine(workflow.EngineOptions{Subagents: sourceNoopSubagents{}}) + manager, err := workflow.NewSourceManager(workflow.SourceManagerOptions{ + Engine: engine, + SkillDirs: []string{filepath.Join(root, "skills")}, + CacheDir: filepath.Join(root, "cache"), + ModuleRoot: mustRepoRoot(t), + }) + require.NoError(t, err) + require.NoError(t, manager.Load(context.Background())) + + var found bool + for _, meta := range manager.List() { + if meta.Name == "skill-review" { + found = true + require.Equal(t, "Skill bundled review workflow.", meta.Description) + require.Equal(t, "Use from the review skill.", meta.WhenToUse) + } + } + require.True(t, found, "expected skill-bundled workflow to be registered") +} + +func newTestSourceManager(t *testing.T, root string, subagents workflow.SubagentManager, questions workflow.QuestionResponder) *workflow.SourceManager { + t.Helper() + engine := workflow.NewEngine(workflow.EngineOptions{Subagents: subagents, QuestionResponder: questions}) + manager, err := workflow.NewSourceManager(workflow.SourceManagerOptions{ + Engine: engine, + WorkflowDirs: []string{filepath.Join(root, "global"), filepath.Join(root, "workspace")}, + SkillDirs: []string{filepath.Join(root, "skills")}, + CacheDir: filepath.Join(root, "cache"), + ModuleRoot: mustRepoRoot(t), + }) + require.NoError(t, err) + return manager +} + +func workflowSourceReturning(expr string) string { + return `package main + +import sdk "go-agent-harness/pkg/workflowsdk" + +func main() { + sdk.Main(func(ctx *sdk.Context) (any, error) { + return ` + expr + `, nil + }) +}` +} + +func runWorkflowToTerminal(t *testing.T, manager *workflow.SourceManager, name string, args map[string]any) *workflow.Run { + t.Helper() + run, err := manager.Start(context.Background(), name, args) + require.NoError(t, err) + waitCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + final, _, err := manager.Wait(waitCtx, run.ID) + require.NoError(t, err) + return final +} + +func mustRepoRoot(t *testing.T) string { + t.Helper() + wd, err := os.Getwd() + require.NoError(t, err) + for { + raw, err := os.ReadFile(filepath.Join(wd, "go.mod")) + if err == nil && strings.Contains(string(raw), "module go-agent-harness") { + return wd + } + parent := filepath.Dir(wd) + require.NotEqual(t, wd, parent, "repo root not found") + wd = parent + } +} diff --git a/internal/workflow/types.go b/internal/workflow/types.go index 634aa49..f306e79 100644 --- a/internal/workflow/types.go +++ b/internal/workflow/types.go @@ -79,10 +79,22 @@ type AgentOpts struct { Schema map[string]any `json:"schema,omitempty"` // Model overrides the model used for this specific agent call. Model string `json:"model,omitempty"` + // Provider overrides the provider used for this specific agent call. + Provider string `json:"provider,omitempty"` + // Profile selects a named sub-agent profile. + Profile string `json:"profile,omitempty"` + // AllowedTools constrains the sub-agent tool set. + AllowedTools []string `json:"allowed_tools,omitempty"` // Isolation selects the isolation mode: "" (inline) or "worktree". Isolation string `json:"isolation,omitempty"` + // CleanupPolicy controls cleanup for isolated sub-agent worktrees. + CleanupPolicy string `json:"cleanup_policy,omitempty"` // AgentType selects a custom sub-agent type (e.g. "Explore", "code-reviewer"). AgentType string `json:"agent_type,omitempty"` + // MaxSteps constrains the child run's maximum step count. + MaxSteps int `json:"max_steps,omitempty"` + // MaxCostUSD constrains the child run's cost ceiling. + MaxCostUSD float64 `json:"max_cost_usd,omitempty"` } // Budget tracks token usage across a workflow run. It is thread-safe. @@ -142,14 +154,18 @@ func newBudget(total int) *Budget { type EventType string const ( - EventWorkflowStarted EventType = "workflow.started" - EventWorkflowPhaseStarted EventType = "workflow.phase.started" - EventWorkflowAgentStarted EventType = "workflow.agent.started" + EventWorkflowStarted EventType = "workflow.started" + EventWorkflowPhaseStarted EventType = "workflow.phase.started" + EventWorkflowAgentStarted EventType = "workflow.agent.started" EventWorkflowAgentCompleted EventType = "workflow.agent.completed" - EventWorkflowAgentFailed EventType = "workflow.agent.failed" - EventWorkflowLog EventType = "workflow.log" - EventWorkflowCompleted EventType = "workflow.completed" - EventWorkflowFailed EventType = "workflow.failed" + EventWorkflowAgentFailed EventType = "workflow.agent.failed" + EventWorkflowLog EventType = "workflow.log" + EventWorkflowFeedback EventType = "workflow.feedback" + EventWorkflowFinding EventType = "workflow.finding" + EventWorkflowWarning EventType = "workflow.warning" + EventWorkflowQuestion EventType = "workflow.question" + EventWorkflowCompleted EventType = "workflow.completed" + EventWorkflowFailed EventType = "workflow.failed" ) // Event is emitted during workflow execution. @@ -163,12 +179,16 @@ type Event struct { // SubagentRequest mirrors the input needed to create a sub-agent run. type SubagentRequest struct { - Prompt string `json:"prompt"` - Model string `json:"model,omitempty"` - Isolation string `json:"isolation,omitempty"` - AgentType string `json:"agent_type,omitempty"` - MaxSteps int `json:"max_steps,omitempty"` - MaxCostUSD float64 `json:"max_cost_usd,omitempty"` + Prompt string `json:"prompt"` + Model string `json:"model,omitempty"` + Provider string `json:"provider,omitempty"` + Profile string `json:"profile,omitempty"` + AllowedTools []string `json:"allowed_tools,omitempty"` + Isolation string `json:"isolation,omitempty"` + CleanupPolicy string `json:"cleanup_policy,omitempty"` + AgentType string `json:"agent_type,omitempty"` + MaxSteps int `json:"max_steps,omitempty"` + MaxCostUSD float64 `json:"max_cost_usd,omitempty"` } // SubagentResult is the result of a completed sub-agent run. @@ -186,6 +206,27 @@ type SubagentManager interface { Get(ctx context.Context, id string) (SubagentResult, error) } +// QuestionOption is a single selectable answer for a workflow question. +type QuestionOption struct { + Label string `json:"label"` + Description string `json:"description"` +} + +// QuestionRequest is emitted when a workflow needs input from the parent/user. +type QuestionRequest struct { + RunID string `json:"run_id"` + CallID string `json:"call_id"` + Prompt string `json:"prompt"` + Choices []QuestionOption `json:"choices,omitempty"` +} + +// QuestionResponder handles workflow questions. Implementations may suspend +// through a checkpoint broker, return a deterministic answer in tests, or +// decline with an error when questions are unavailable. +type QuestionResponder interface { + AskWorkflowQuestion(ctx context.Context, req QuestionRequest) (any, error) +} + // PipelineStage is a function that processes one item through one stage. // prev is the result from the previous stage (nil for the first stage). // item is the original input item. diff --git a/pkg/workflowsdk/sdk.go b/pkg/workflowsdk/sdk.go new file mode 100644 index 0000000..892758d --- /dev/null +++ b/pkg/workflowsdk/sdk.go @@ -0,0 +1,237 @@ +package workflowsdk + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "sync" + "sync/atomic" +) + +// Run is the entrypoint signature for Go-authored workflows. +type Run func(ctx *Context) (any, error) + +// Context exposes host-backed workflow operations to dynamically compiled +// workflow binaries. +type Context struct { + Args any + + client *client +} + +type AgentOpts struct { + Label string `json:"label,omitempty"` + Phase string `json:"phase,omitempty"` + Schema any `json:"schema,omitempty"` + Model string `json:"model,omitempty"` + Provider string `json:"provider,omitempty"` + Profile string `json:"profile,omitempty"` + AllowedTools []string `json:"allowed_tools,omitempty"` + Isolation string `json:"isolation,omitempty"` + CleanupPolicy string `json:"cleanup_policy,omitempty"` + AgentType string `json:"agent_type,omitempty"` + MaxSteps int `json:"max_steps,omitempty"` + MaxCostUSD float64 `json:"max_cost_usd,omitempty"` +} + +type AgentResult struct { + Output string `json:"output"` + Schema any `json:"schema,omitempty"` + Error string `json:"error,omitempty"` +} + +type QuestionOption struct { + Label string `json:"label"` + Description string `json:"description"` +} + +// Main connects a workflow binary to the host JSONL protocol and runs fn. +func Main(fn Run) { + c, args, err := newClient(os.Stdin, os.Stdout) + if err != nil { + writeBootError(os.Stdout, err) + os.Exit(1) + } + ctx := &Context{Args: args, client: c} + result, runErr := fn(ctx) + if runErr != nil { + _ = c.sendTerminal("error", nil, runErr) + os.Exit(1) + } + if err := c.sendTerminal("result", result, nil); err != nil { + os.Exit(1) + } +} + +func (c *Context) Agent(prompt string, opts *AgentOpts) (*AgentResult, error) { + var out AgentResult + err := c.client.call("agent", map[string]any{ + "prompt": prompt, + "opts": opts, + }, &out) + if err != nil { + return nil, err + } + return &out, nil +} + +func (c *Context) Phase(title string) error { + return c.client.call("phase", map[string]any{"title": title}, nil) +} + +func (c *Context) Log(message string) error { + return c.client.call("log", map[string]any{"message": message}, nil) +} + +func (c *Context) Feedback(kind, message string, data map[string]any) error { + return c.client.call("feedback", map[string]any{ + "kind": kind, + "message": message, + "data": data, + }, nil) +} + +func (c *Context) Question(prompt string, choices []QuestionOption) (any, error) { + var out any + err := c.client.call("question", map[string]any{ + "prompt": prompt, + "choices": choices, + }, &out) + return out, err +} + +func (c *Context) Workflow(name string, args any) (any, error) { + var out any + err := c.client.call("workflow", map[string]any{ + "name": name, + "args": args, + }, &out) + return out, err +} + +type client struct { + enc *json.Encoder + encMu sync.Mutex + nextID atomic.Int64 + + pendingMu sync.Mutex + pending map[string]chan response +} + +type message struct { + ID string `json:"id,omitempty"` + Type string `json:"type"` + Args any `json:"args,omitempty"` + Result any `json:"result,omitempty"` + Error string `json:"error,omitempty"` + Raw json.RawMessage `json:"-"` +} + +type response struct { + ID string `json:"id,omitempty"` + Type string `json:"type"` + Result json.RawMessage `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +func newClient(stdin io.Reader, stdout io.Writer) (*client, any, error) { + dec := json.NewDecoder(bufio.NewReader(stdin)) + var start response + if err := dec.Decode(&start); err != nil { + return nil, nil, fmt.Errorf("read workflow start message: %w", err) + } + if start.Type != "start" { + return nil, nil, fmt.Errorf("expected start message, got %q", start.Type) + } + var args any + if len(start.Result) > 0 { + if err := json.Unmarshal(start.Result, &args); err != nil { + return nil, nil, fmt.Errorf("decode workflow args: %w", err) + } + } + c := &client{ + enc: json.NewEncoder(stdout), + pending: make(map[string]chan response), + } + go c.readLoop(dec) + return c, args, nil +} + +func (c *client) call(typ string, args any, out any) error { + id := fmt.Sprintf("req_%d", c.nextID.Add(1)) + ch := make(chan response, 1) + c.pendingMu.Lock() + c.pending[id] = ch + c.pendingMu.Unlock() + + c.encMu.Lock() + err := c.enc.Encode(message{ID: id, Type: typ, Args: args}) + c.encMu.Unlock() + if err != nil { + c.dropPending(id) + return err + } + + resp := <-ch + if resp.Error != "" { + return errors.New(resp.Error) + } + if out != nil && len(resp.Result) > 0 { + if err := json.Unmarshal(resp.Result, out); err != nil { + return err + } + } + return nil +} + +func (c *client) sendTerminal(typ string, result any, err error) error { + msg := message{Type: typ, Result: result} + if err != nil { + msg.Error = err.Error() + } + c.encMu.Lock() + defer c.encMu.Unlock() + return c.enc.Encode(msg) +} + +func (c *client) readLoop(dec *json.Decoder) { + for { + var resp response + if err := dec.Decode(&resp); err != nil { + c.failAll(err) + return + } + if resp.ID == "" { + continue + } + c.pendingMu.Lock() + ch := c.pending[resp.ID] + delete(c.pending, resp.ID) + c.pendingMu.Unlock() + if ch != nil { + ch <- resp + } + } +} + +func (c *client) dropPending(id string) { + c.pendingMu.Lock() + delete(c.pending, id) + c.pendingMu.Unlock() +} + +func (c *client) failAll(err error) { + c.pendingMu.Lock() + defer c.pendingMu.Unlock() + for id, ch := range c.pending { + delete(c.pending, id) + ch <- response{ID: id, Error: err.Error()} + } +} + +func writeBootError(stdout io.Writer, err error) { + _ = json.NewEncoder(stdout).Encode(message{Type: "error", Error: err.Error()}) +}