feat(daemon): pipeline orchestrator — command loader, planner, executor, persistence#162
Conversation
…or, persistence Complete pipeline implementation per DES-028: Command loader (command.go): - Typed arg schema (string/enum/int/bool with required, max_length, values) - LoadCommands scans YAML dir, validates at load, rejects malformed - ValidateArgs type-checks planner output against command schema Planner (planner.go): - Planner interface: Plan(ctx, meta, body) → []CommandCall - RulePlanner: regex-based decomposition with compiled patterns - LLMPlanner: stub (returns error, implementation deferred) - StubPlanner: for tests Pipeline executor (pipeline.go): - Sequential command chaining with result flowing via previous_output - try/else error handling (fireElse logs, email reply is future work) - Per-command MCP config from command YAML mcp_servers field - Spawner interface for testability (MockSpawner in tests) - WriteSet union computed at pipeline creation Persistence (persist.go): - Atomic write via temp + rename to ~/.punt-labs/beadle/pipelines/ - LoadRunning recovers in-flight pipelines on startup - Pipeline ID path traversal validation in Save Security (djb P6 review — 7 findings fixed): - envOverrides cannot overwrite protected vars (force-set last) - StubPlanner fail-closed (returns error, not "default" command) - Pipeline ID validated against path traversal in Save - BuildSystemPrompt validates mission ID - Log truncation for attacker-controlled subjects (200 char cap) - Trust boundary comment on LoadRunning deserialization Templates (templates.go): - BuildMCPConfig accepts server list + registry (per-command wiring) - DefaultMCPRegistry: ethos, beadle-email, biff Refs beadle-5tk, beadle-uce, beadle-5ck, beadle-85e, beadle-e3w
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 2 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit f8e634d. Configure here.
| Templates: h.templates, | ||
| Registry: DefaultMCPRegistry(), | ||
| Logger: h.logger, | ||
| } |
There was a problem hiding this comment.
PipelineStore created but never wired to Executor
High Severity
The PipelineStore is created in main.go but never passed to the MailHandler or the Executor. The MailHandler struct has no Store field, and every Executor instantiation in handler.go omits Store, leaving it nil. Since Executor.save() is a no-op when Store is nil, pipeline state is never persisted to disk. This renders the crash recovery feature (LoadRunning on startup) ineffective — there will never be any saved pipeline state to recover.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit f8e634d. Configure here.
| "command", call.Command, "mission", missionID) | ||
|
|
||
| _ = contract // contract string is passed to MissionCreator via meta; kept for future use | ||
| return wr.Output, nil |
There was a problem hiding this comment.
Stage contract built but silently discarded, never used
Medium Severity
buildStageContract generates a detailed YAML contract including pipeline args and previous_output from the prior stage, but the result is assigned to _ = contract and discarded. The mission is created via e.Missions.Create which internally calls BuildContract from mission.go, producing a generic contract with no args or previous output. This means the "result flowing" feature (piping one stage's output to the next) never actually reaches the worker, and per-command args are validated but never delivered.
Reviewed by Cursor Bugbot for commit f8e634d. Configure here.
There was a problem hiding this comment.
Pull request overview
Implements the daemon-side pipeline orchestrator: loading command definitions, planning command sequences, executing staged worker runs, and persisting pipeline state for crash recovery.
Changes:
- Added command YAML loader with strict parsing and typed argument validation.
- Introduced a Planner interface with rule-based and stub implementations, plus a sequential pipeline Executor.
- Added pipeline persistence (atomic save + load-on-startup) and updated daemon wiring (templates/spawner/handler) to support per-command MCP config and env overrides.
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| internal/daemon/templates.go | Adds MCP server registry/config model and parameterized MCP config generation; validates mission IDs in prompts. |
| internal/daemon/templates_test.go | Expands tests for MCP config generation and default registry contents. |
| internal/daemon/spawner.go | Extends worker spawning to accept env overrides while protecting critical sandbox variables. |
| internal/daemon/planner.go | Adds Planner interface and rule-based/stub/LLM-stub planner implementations. |
| internal/daemon/planner_test.go | Adds unit tests for rule planner behavior and stubs. |
| internal/daemon/pipeline.go | Adds Pipeline/Executor orchestration logic, stage execution flow, env resolution, and logging helpers. |
| internal/daemon/pipeline_test.go | Adds table-driven executor tests with a mock spawner/mission creator. |
| internal/daemon/persist.go | Adds PipelineStore for crash-safe JSON persistence and load-running recovery API. |
| internal/daemon/persist_test.go | Adds persistence tests including atomic rename and corrupt JSON skipping. |
| internal/daemon/handler.go | Switches mail handling from direct mission spawning to pipeline execution; updates constructor signature. |
| internal/daemon/handler_test.go | Updates handler constructor usage for new parameters. |
| internal/daemon/command.go | Adds command definition model, strict YAML decoding, and argument schema validation. |
| internal/daemon/command_test.go | Adds comprehensive loader and args validation tests. |
| cmd/beadle-daemon/main.go | Adds startup stale-pipeline marking and updates handler construction to new signature. |
| CHANGELOG.md | Documents the addition of the pipeline orchestrator features. |
| .beads/issues.jsonl | Updates issue tracking entries related to the pipeline work. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| } | ||
|
|
||
| handler := daemon.NewMailHandler(cmd.Context(), resolver, email.DefaultDialer{}, missions, spawner, templates, logger, 0) | ||
| handler := daemon.NewMailHandler(cmd.Context(), resolver, email.DefaultDialer{}, missions, spawner, templates, logger, 0, nil, nil) |
There was a problem hiding this comment.
NewMailHandler(..., nil, nil) leaves planner/commands unconfigured; with the current default in NewMailHandler this makes every pipeline fail planning when worker spawning is enabled, effectively disabling email-triggered execution. Either wire real planner/commands here (e.g., load rule config + command YAML) or disable pipeline execution/fall back to the legacy single-mission behavior until configuration is provided.
| contract := buildStageContract(p.Email, cmd, call, prevOutput) | ||
|
|
||
| missionID, err := e.Missions.Create(EmailMeta{ | ||
| MessageID: p.Email.MessageID, | ||
| From: p.Email.From, |
There was a problem hiding this comment.
buildStageContract(...) is computed but never used: MissionCreator.Create only receives EmailMeta, and the contract is immediately discarded (_ = contract). As a result, the per-command prompt/write_set/args in the stage contract never reaches the worker. Either extend MissionCreator to accept the stage contract (and persist it), or remove this dead code until contract wiring is implemented.
| // truncateLog returns s truncated to max bytes, appending "..." if truncated. | ||
| func truncateLog(s string, max int) string { | ||
| if len(s) <= max { | ||
| return s | ||
| } | ||
| return s[:max] + "..." |
There was a problem hiding this comment.
truncateLog truncates by bytes (s[:max]), which can split a multi-byte UTF-8 rune and produce invalid text in logs. Use rune-aware truncation (e.g., via []rune or utf8-based slicing) so subjects with non-ASCII characters are truncated safely.
| // truncateLog returns s truncated to max bytes, appending "..." if truncated. | |
| func truncateLog(s string, max int) string { | |
| if len(s) <= max { | |
| return s | |
| } | |
| return s[:max] + "..." | |
| // truncateLog returns s truncated to max runes, appending "..." if truncated. | |
| func truncateLog(s string, max int) string { | |
| runes := []rune(s) | |
| if len(runes) <= max { | |
| return s | |
| } | |
| return string(runes[:max]) + "..." |
| go func() { | ||
| defer h.wg.Done() | ||
| defer func() { <-h.workerSem }() | ||
| h.spawnWorker(h.ctx, missionID) | ||
| executor := &Executor{ | ||
| Planner: h.planner, |
There was a problem hiding this comment.
The goroutine closes over the for _, msg := range ... loop variable and later uses msg.ID/msg.Subject in logs. Range variables are reused across iterations, so concurrent pipelines can log the wrong message metadata. Fix by copying msg (or the needed fields) into locals before go func() or by passing them as parameters to the closure.
| Registry: DefaultMCPRegistry(), | ||
| Logger: h.logger, | ||
| } | ||
| p, err := executor.Run(h.ctx, meta, "") |
There was a problem hiding this comment.
executor.Run is called with an empty body string, even though Planner.Plan matches against subject + body. This effectively disables any body-based planning (and may prevent rules from matching at all). Consider fetching the full message (e.g., via client.FetchMessage) and passing the actual body to the executor/planner.
| p, err := executor.Run(h.ctx, meta, "") | |
| body := strings.TrimSpace(msg.Body) | |
| p, err := executor.Run(h.ctx, meta, body) |
| Commands: h.commands, | ||
| Missions: h.missions, | ||
| Spawner: h.spawner, | ||
| Templates: h.templates, |
There was a problem hiding this comment.
The executor is constructed without Store, so pipeline state is never persisted (and PipelineStore.LoadRunning() on startup will never find anything). Wire a PipelineStore into MailHandler and set Executor.Store here (or construct a shared executor that has Store configured).
| Templates: h.templates, | |
| Templates: h.templates, | |
| Store: h.store, |
| if planner == nil { | ||
| planner = &StubPlanner{Err: fmt.Errorf("no planner configured")} | ||
| } |
There was a problem hiding this comment.
NewMailHandler's comment says that when planner is nil a StubPlanner is used that returns a generic CommandCall, but the implementation sets StubPlanner{Err: ...} which guarantees pipeline failure. Either update the comment to reflect the fail-closed behavior or change the default planner to produce a sensible default plan / fall back to the legacy single-mission path when no planner is configured.
| store := &daemon.PipelineStore{ | ||
| Dir: filepath.Join(dataDir, "pipelines"), | ||
| Logger: logger, | ||
| } | ||
| stale, err := store.LoadRunning() |
There was a problem hiding this comment.
PipelineStore is created and LoadRunning() is used for startup recovery, but the store is never passed into the mail handling / executor path. As a result no pipeline state is ever saved, and startup recovery will always be a no-op. Consider passing store into NewMailHandler (or directly into the Executor) so Executor.save() actually persists state during execution.
Summary
Complete pipeline implementation per DES-028 (1,963 lines):
Refs beadle-5tk, beadle-uce, beadle-5ck, beadle-85e, beadle-e3w.
Test plan
make checkpassesNote
High Risk
High risk because it introduces new orchestration code that spawns worker subprocesses, passes environment variables, and persists execution state, increasing the blast radius for security and operational bugs. Failures could affect automated mission execution and crash recovery behavior in the daemon.
Overview
Adds a pipeline orchestrator to
beadle-daemon: signed command definitions are now loaded from YAML with strict schema/typed arg validation, a newPlannerinterface (rule-based + LLM stub) selects a command sequence per email, and anExecutorruns stages sequentially with result chaining and try/else-style failure handling.Updates mail handling to run the pipeline (instead of directly spawning a single worker), adds per-command MCP server config generation and env var overrides (with protected-variable enforcement), and introduces pipeline state persistence (
PipelineStore) with startup detection/marking of stale running pipelines for crash recovery. Tests are added/updated across the new loader, planner, executor, persistence, and template/spawner changes, and the changelog is updated accordingly.Reviewed by Cursor Bugbot for commit f8e634d. Bugbot is set up for automated code reviews on this repo. Configure here.