DarkFactory engine — technical map

State: open/active · Genre: plan

Companion to README.md. A full map of the DarkFactory Python harness, read with one question in mind: how does it work as an engine that any agent can drive, and how does it call the agent? All paths are under python/darkfactory/ unless noted. Python ≥3.12, mypy --strict, runtime deps: pyyaml, filelock, rich.

Summary

DarkFactory is a deterministic engine that runs a ticket through a workflow and treats an LLM agent as one step type out of four. A PRD (a markdown ticket with YAML frontmatter) runs through a Workflow (an ordered list[Task]). The engine walks that list and dispatches each step by type: BuiltIn (deterministic Python — worktree/commit/push/PR/status), ShellTask (a check like just test), AgentTask (headless Claude), and InteractiveTask (interactive Claude). The harness — never the agent — owns all git/GitHub changes, all branching, and all status changes. The agent’s whole job is: write code, stage it, print one line. That line (PRD_EXECUTE_OK: / PRD_EXECUTE_FAILED:) is the entire pass/fail contract.

How the agent is called (the key part): a subprocess to the Claude Code CLI — pnpm dlx @anthropic-ai/claude-code --print --verbose --output-format stream-json --model <m> --allowed-tools <csv> --disallowed-tools Edit(../) …, prompt piped in over stdin, stream-json parsed line by line, pass/fail decided by matching a regex on a line of stdout (with an exit-code cross-check). No SDK. No MCP anywhere. No structured-output/tool-call contract — tools are just a Claude Code allowlist string.

Any-agent vs Claude-specific: the engine (DAG, PhaseState, workflows, builtins, runner, locking/events) does not know about Claude. Claude-specific code sits in about two files (utils/claude_code/), plus the Claude Code tool names in each workflow’s tools=[...] list and the model-alias map. There is no abstract Agent interface — the seam is one isolated function (invoke_claude) with an existing _argv_override test hook. The reusable idea: the harness owns all changes, the agent only writes/stages/signals, enforced by tool allowlists instead of trust.

1. Engine architecture — following `prd run <id>`

Entry: darkfactory.cli:main (console scripts prd/darkfactory, pyproject.toml:14-15).

cli/main.py:39-81 — builds the argparse tree, sets up stderr logging (_configure_logging — this matters, since streamed agent output goes through logging.info), finds the project root by walking up for .darkfactory/, sets args.data_dir, resolves config + Styler, then return int(func(args)).
cli/run.py:69-191 — cmd_run picks one of three modes:
- Queue (--all): QueueFilters + QueueStrategy → _cmd_run_queue → graph executor.
- Graph: _is_graph_target (run.py:50-66, true when the PRD has children or unfinished depends_on) → _cmd_run_graph → graph executor.
- Legacy single-PRD: a ready leaf with all deps done → resolve workflow → run_workflow directly (run.py:158-169).
- Default is dry-run; --execute opts in (run.py:95); session_id is generated only on execute.
runner.py:619-701 — run_workflow is the single-PRD path: it computes branch_name = compute_branch_name(prd) (prd/{id}-{slug}), optionally makes an EventWriter, builds RunContext, seeds PhaseState with CodeEnv(cwd=repo_root), PrdWorkflowRun(prd, workflow), WorktreeState(branch, base_ref), then calls run_tasks(...) with BUILTINS, _workflow_compose_prompt, capability-aware model picking, and timeouts from prd.effort/prd.capability/raw_frontmatter["timeout_minutes"]. Finally it copies PrResult.url into RunResult.pr_url.
run_workflow and run_project_operation (runner.py:734-796) both go through one dispatch function, run_tasks, which takes the builtin registry, the prompt composer, and the model picker as parameters. This is the cleanest seam in the code.
runner.py:148-276 — run_tasks walks steps in order and dispatches by isinstance:
- BuiltIn → _run_builtin (:282-306): look up the name, format string kwargs via ctx.format_string, call func(ctx, **kwargs).
- AgentTask → _run_agent (:309-381): compose the prompt, pick the model, resolve the timeout, invoke_claude(...), store AgentResult in PhaseState.
- ShellTask → _run_shell (:384-514): run the command, apply the on_failure policy.
- InteractiveTask → _run_interactive (:517-569): spawn_claude.
Stop-on-failure, sequential: the loop stops on the first step that returns success=False or raises (caught at :254-274, recorded as a failed TaskStep, then break). RunResult = success, pr_url, steps, failure_reason.

2. The DAG model

Two separate graphs come from frontmatter:

Dependency DAG (graph/_dag.py): Graph = dict[str, set[str]], edges dep → prd. detect_cycles = Tarjan SCC (:42-86). topological_sort = Kahn with a natural-number tie-break via parse_id_sort_key (:89-120). is_actionable = ready + all deps done (:142-150).
Containment tree (graph/_containment.py): the parent field. is_leaf, is_fully_decomposed (has a task-kind descendant, :61-67), and is_runnable (:88-97) — runnable if kind=="task" OR it is a leaf. Epics/features with children are not runnable and route to planning.
Scheduling/execution (graph/_execution.py, sequential only): graph_scope (root ∪ descendants ∪ transitive unmet-dep closure), actionable_order (topological sort of runnable PRDs), a CandidateStrategy Protocol with RootedStrategy/QueueStrategy. execute_graph (:441-667) loops: load_all(data_dir) on every pass (so planning can grow the DAG mid-run and queue mode picks up newly-ready PRDs), computes effective_skipped (sticky-skip ∪ blocked ∪ transitive dependents — this isolates failures), picks the first runnable candidate, resolves the base ref (resolve_base_ref: one dep finished this run → stack on its branch; ≥2 unmerged → MultiDepUnsupported skip), runs the workflow, and on failure adds to blocked_ids and set_status_at(path, "blocked"). State across nodes is in-memory dicts (completed_this_run, sticky_skipped, blocked_ids) — not persisted.
State within a run: engine/phase_state.py — a store keyed by type (_store: dict[type, Any]); each value’s type is its key, and mypy infers get(T) -> T via @overload. Payloads (engine/payloads.py, all @dataclass(frozen=True)): CodeEnv(repo_root, cwd) (replaced by ensure_worktree so cwd points at the worktree), PrdWorkflowRun, ProjectRun, WorktreeState, PrRequest, PrResult, PrdContext, AgentResult(stdout, stderr, exit_code, success, failure_reason, tool_counts, sentinel, model, invoke_count), CandidateList, ReworkState. Payloads are immutable and replaced, not mutated — a clean message bus between steps.

3. Workflow definitions — the declarative layer

Step types (workflow/_core.py): Task (marker base); BuiltIn(name, kwargs); AgentTask(name, prompts, tools, model, model_from_capability, retries, verify_prompts, sentinel_success, sentinel_failure, effort_level); ShellTask(name, cmd, on_failure∈{fail,retry_agent,ignore}, env); InteractiveTask(name, prompt_file, tools, model, effort_level). Workflow(name, description, applies_to, priority, tasks, workflow_dir, template_name). RunContext carries dry_run, logger, state, event_writer, report, _worktree_lock, plus format_string, which expands {lowercase} placeholders from payloads (separate from the agent-prompt {{UPPERCASE}} system).

Concrete example — default (definitions/prd/default/workflow.py, the catchall, applies_to True, priority=0):

workflow = PRD_IMPLEMENTATION_TEMPLATE.compose(
    name="default",
    applies_to=_applies_to_everything,
    priority=0,
    middle=[
        AgentTask(
            name="implement",
            prompts=["prompts/role.md", "prompts/task.md"],
            tools=["Read","Edit","Write","Glob","Grep",
                   "Bash(cargo:*)","Bash(pnpm:*)","Bash(just:*)","Bash(uv:*)",
                   "Bash(git add:*)","Bash(git status:*)","Bash(git diff:*)","Bash(git log:*)"],
            model_from_capability=True, retries=1,
            verify_prompts=["prompts/verify.md"],
        ),
        ShellTask("test", cmd="just test", on_failure="retry_agent"),
        ShellTask("format", cmd="just format", on_failure="fail"),
        ShellTask("lint", cmd="just lint format-check", on_failure="retry_agent"),
        ShellTask("typecheck", cmd="just typecheck", on_failure="retry_agent"),
    ],
)

The agent allowlist includes only git add/status/diff/log — no commit, push, or gh.

Templates (workflow/_templates.py, _templates_builtin.py): WorkflowTemplate defines the required open builtins, a constrained middle (checked by middle_kinds/middle_required count bounds), and the required close; .compose() returns Workflow(tasks=[*open,*middle,*close]). PRD_IMPLEMENTATION_TEMPLATE open = ensure_worktree, set_status{in-progress}, commit{start work}; close = set_status{review}, commit_events, analyze_transcript, commit{ready for review}, summarize_agent_run, lint_attribution, push_branch, create_pr. Four templates ship: PRD_IMPLEMENTATION, EXTRACTION (agent-only middle, separate target repo, no just steps), SYSTEM_OPERATION (lock→execute→report→unlock), REWORK.

Other workflows: PRD: default, planning, planning_review, extraction, task. Project: plan, verify_merges, audit_impacts.

planning breaks up epics: the predicate is epic/feature + ready + not-fully-decomposed; model="opus"; a narrower allowlist (Read/Glob/Grep/Write/Edit, Bash(uv run prd validate*), git add scoped to .darkfactory/prds/); it writes child PRD files, updates the parent’s blocks: wikilinks, and runs uv run prd validate. planning_review (priority 6 > 5) handles partly-decomposed epics and is Write-only (no Edit).
verify_merges is a ShellTask-only workflow (no agent) that runs check.py — it lists merged PRs via gh and asserts git merge-base --is-ancestor <mergeCommit> main.
audit_impacts is a BuiltIn-only workflow — a single BuiltIn("audit_impacts_check") that checks the declared impacts: paths exist.

Extensibility: loader.py:57-132 load_workflows scans two layers (built-in workflow/definitions/prd/, project <repo>/.darkfactory/workflows/) for <name>/workflow.py exporting workflow = Workflow(...), imports via importlib, sets workflow_dir, and hard-fails on cross-layer name collisions. load_project_workflows adds a user layer (~/.config/darkfactory/operations/). A project adds a workflow by dropping in a Python file — no DarkFactory code changes. Builtins, by contrast, are a fixed registry inside the package.

4. Operations / built-in steps — the contract

operations/_registry.py: BuiltInFunc = Callable[..., None] (signature (RunContext, **kwargs) -> None); the BUILTINS dict is filled by @builtin("name") decorators. Contract: every builtin takes (ctx, **kwargs), returns None, reports results by replacing PhaseState payloads, and signals failure by raising. operations/__init__.py imports every submodule so the decorators run at import. A parallel SYSTEM_BUILTINS (operations/project_builtins.py) serves project workflows; git ops are copied into both registries.

Idempotency and side effects, worked examples:

ensure_worktree (:62-159): dry-run swaps payloads; live mode takes a non-blocking FileLock(".worktrees/{prd_id}.lock") before any change, then resumes an existing worktree if is_resume_safe (this is what makes re-running prd run idempotent), else git worktree add.
set_status (:14-43): rewrites the frontmatter in the worktree copy via set_status_at, mirrors it onto the in-memory PRD.
commit (:13-63): git add -A, git diff --cached --quiet; empty diff → log and return (so it is safe to commit after every step); scans the message for forbidden attribution.
create_pr (:61-153): title/body cascade (kwargs > PrRequest > PRD defaults
- AC checklist), writes the body to a temp file, runs gh pr create, stores PrResult; dry-run stores a fake URL.

Pass/fail/retry:

BuiltIn: success = returns; failure = raises (stops the workflow).
AgentTask: decided by sentinel parsing + an exit-code cross-check. The retries field is not a general agent retry; retry happens via ShellTask retry_agent (runner.py:448-514): a failed shell step makes a one-shot retry AgentTask from the previous agent using its verify_prompts, binds the failed output to {{CHECK_OUTPUT}}, calls the agent once more, and re-runs the command once.
ShellTask on_failure: fail (stop), retry_agent (above), ignore (log and continue).

5. How the agent is called (the key part)

All in utils/claude_code/, two files. It is a subprocess to the Claude Code CLI — not the SDK, not MCP.

_invoke.py:8-15 docstring: “Subprocess, not SDK: we shell out to the claude CLI (via pnpm dlx @anthropic-ai/claude-code)…”

The command it builds (_invoke.py:415-436):

cmd = ["pnpm", "dlx", "@anthropic-ai/claude-code",
       "--print", "--verbose", "--output-format", "stream-json",
       "--model", model, "--allowed-tools", ",".join(tools),
       "--disallowed-tools", "Edit(../)",
       "--disallowed-tools", "Write(../)",
       "--disallowed-tools", "Read(../)"]
if effort_level is not None: cmd.extend(["--effort", effort_level])

The prompt is piped in over stdin (:457-459); cwd = the worktree (kept inside it by the (../) disallows). No MCP anywhere — no MCP server/config/--mcp-config/client. Tools are Claude Code’s own tool-allowlist names.

What goes in: the prompt (built by compose_prompt, _templates.py:104-154 — it concatenates the AgentTask.prompts files and substitutes {{PRD_ID}}/{{PRD_TITLE}}/{{WORKTREE_PATH}}/{{BRANCH_NAME}}/ {{REWORK_FEEDBACK}}/{{CHECK_OUTPUT}} etc.; no system prompt, no tool schemas, no history — one string to a one-shot --print), the tools (allowlist csv), the model (_pick_model runner.py:91-102: --model > task.model > capability_to_model(prd.capability) > sonnet; map trivial→haiku, simple/moderate→sonnet, complex→opus), and the timeout (resolve_timeout).

How output is read: invoke_claude (_invoke.py:352-618) reads the stream-json line by line; _summarize_stream_event (:205-346) classifies events and builds up the assistant text so it can scan for the sentinel; a watchdog thread enforces the wall-clock timeout (proc.kill()); tool-use events increment tool_counts; raw events are written to the per-PRD JSONL event log.

The structured-output contract is a sentinel line (_invoke.py:132-133):

_SENTINEL_SUCCESS_RE = re.compile(r"PRD_EXECUTE_OK:\s*(\S[^\n`]*)")
_SENTINEL_FAILURE_RE = re.compile(r"PRD_EXECUTE_FAILED:\s*(\S[^\n`]*)")

_parse_sentinels (:136-199) filters out darkfactory_ envelope lines, then matches; failure beats success; matching is anchorless (agents wrap sentinels in markdown). Verdict assembly (:582-618): if exit_code != 0 but a success sentinel is present → downgraded to failure; on timeout, honor a terminal {"type":"result"} event verdict if present. It returns InvokeResult; the runner copies it into the AgentResult payload.

The agent is told the contract in definitions/prd/default/prompts/role.md: “Your final line of output must be exactly one of: PRD_EXECUTE_OK: {{PRD_ID}} … PRD_EXECUTE_FAILED: <reason> … No other output format will be recognized.” And: “You are one link in a larger harness … your job is the implementation and staging step only … You MUST NOT: Run git commit … Create or switch git branches … Push to origin or create pull requests … Invoke other agents.”

Other modes: spawn_claude (_interactive.py:21-38) — interactive, bare claude --effort <level> <prompt>, inherits the terminal, returns the exit code (used by InteractiveTask/discuss). claude_print (:41-72) — headless pnpm dlx … --print, captured stdout (used by analyze_transcript as a Haiku/Sonnet retro narrator with --allowed-tools Read).

Is it behind an interface? Partly. There is no formal Agent protocol/ABC; the runner calls the concrete invoke_claude at one site (runner.py:339-352). But the seam is narrow — one function, a clean InvokeResult return, plus an existing _argv_override test hook (_invoke.py:413-414). To swap agents you reimplement invoke_claude + _summarize_stream_event; nothing in the runner, workflow types, DAG, or builtins changes.

6. State and concurrency

.darkfactory/ layout (checked on disk): config.toml; data/prds/ (active PRDs — the source of truth) + data/archive/; events/PRD-<id>-<ts>.jsonl (append-only structured log); state/rework-guard.json + rework-watch.json; transcripts/; workflows/ + operations/ (project-layer modules). The actual git worktrees live at repo-root .worktrees/ (not under .darkfactory/) with {id}.lock files; a global lock sits at .harness-system.lock.

PRD writes are not atomic — both save (model/_persistence.py:287-299) and the surgical set_status_at/update_frontmatter_field_at (:310-369) use plain path.write_text (no temp file + os.replace, no fsync). Durability under a crash relies on the file locks, not on atomic writes.

Event log (event_log.py): EventWriter writes one JSONL line per event, flushes each write, and guards it with a per-writer threading.Lock; the envelope is ts/session_id/prd_id/scope/type; the flat-field shape is meant for future parallelism.

filelock concurrency — three scopes: (1) a per-worktree/branch FileLock(".worktrees/{prd_id}.lock"), non-blocking, held on ctx._worktree_lock, released in the runner’s finally — this serializes two prd runs on the same PRD; (2) a global FileLock(".harness-system.lock"), non-blocking — this serializes all project operations repo-wide; (3) the in-process EventWriter thread lock.

Crash recovery: there is no transactional snapshot. It is rebuilt from durable artifacts — PRD status flips persist; re-running prd run resumes the existing worktree if is_resume_safe (checks.py:61-137: refuses only on a MERGED/CLOSED PR; divergence is a warning, not a stop); idempotent builtins make replay mostly safe (the agent step just re-runs from scratch); execute_graph re-runs load_all each pass, so persisted status is picked up.

7. Status model and transitions

Status enum (workflow/_core.py:64-73): draft, ready, in-progress, review, done, blocked, cancelled, superseded. The deterministic engine drives transitions, not the agent — they are hard-coded as BuiltIn("set_status", …) template steps: ready→in-progress (open list, before the agent), in-progress→review (close list, after the checks), in-progress→blocked (set by execute_graph on failure). review→done is manual/out-of-band — only system_mark_done flips it after system_check_merged confirms a merged branch (the vision doc lists auto-review→done as an open question). Decomposition routing (epic→planning) is also deterministic, via the is_runnable/ is_fully_decomposed predicates. The agent never decides status.

8. Rework / failure handling

Two mechanisms:

In-workflow shell failure → agent retry (runner.py:384-514, the retry_agent policy): makes a one-shot retry AgentTask from the previous agent with its verify_prompts and {{CHECK_OUTPUT}} set to the failure, calls the agent again, and re-runs the command once. One retry per shell step.
The rework/ package — the post-PR “changes requested” cycle, started by prd rework <id> --execute (definitions/prd/rework/workflow.py). Unlike default, it does not create a worktree, change status, or open a PR — it reuses the existing ones. rework/context.py discover_rework_context finds the worktree, finds the open PR (gh pr list --head <branch> --state open), checks the guard, and fetches unresolved review threads via GitHub GraphQL (gh api graphql, pulling reviewThreads + reviews + comments in one call). rework/prompt.py renders the threads into the {{REWORK_FEEDBACK}} block. rework/guard.py ReworkGuard does loop detection — it keeps per-PRD consecutive-no-change counters in .darkfactory/state/rework-guard.json; after 2 no-change cycles the PRD is blocked until a manual reset (the check_rework_guard builtin raises if blocked). The workflow: resolve_rework_context → fast_forward_branch → rebase_onto_main → AgentTask(effort_level="max") → format/lint/typecheck/test → check_rework_guard → commit → push_branch (auto-updates the PR) → optional reply_pr_comments (REST replies of the form [harness] addressed in <sha>: …).

9. Dependencies

requires-python = ">=3.12"; dependencies = ["pyyaml>=6.0","filelock>=3.13", "rich>=13.0"]. Three runtime deps. No Anthropic SDK, no HTTP client, no MCP library, no database, no async framework. The agent is reached only by shelling out to pnpm dlx @anthropic-ai/claude-code (so pnpm + the Claude Code npm package are environment prerequisites, not Python deps) and to git/gh. Everything external is subprocess.run/Popen with argv lists, never shell=True except the one deliberate run_shell for workflow commands. The git/GitHub plumbing (utils/git/, utils/github/) is a small layer over a dependency-free Result type (Ok/GitErr/GhErr/Timeout frozen dataclasses, match/case at call sites, “never raise, return a typed result”).

10. Any-agent vs Claude-specific

Any-agent (reusable engine): the DAG + scheduling (graph/* — plain Tarjan/Kahn over frontmatter, no LLM concept); the PhaseState typed bus (engine/*); the workflow/step types (workflow/_core.py — AgentTask is just data: prompt + tool list + model string); the dispatch engine (runner.py run_tasks, parameterized, one invoke_claude call site); the builtins (deterministic git/gh); state/locking/events; the workflow loader. The orchestration half knows nothing about Claude.

Claude-specific (all isolated): (1) utils/claude_code/_invoke.py — the CLI command, the flags (--print/--verbose/--output-format stream-json/--model/ --allowed-tools/--disallowed-tools/--effort), the stream-json event shape parsing, timeout/terminal-result handling. (2) _interactive.py — spawn_claude/claude_print. (3) the capability→model map (haiku/sonnet/opus) and the --effort levels. (4) Tool names in every AgentTask.tools list (Read, Edit, Bash(just:*), the Edit(../) disallows) — Claude Code’s permission syntax appears in each workflow definition. (5) The sentinel contract is agent-neutral in idea, but the teaching prompts + the parser are part of it. (6) No MCP/SDK/tool-schema layer — agent capabilities are exactly Claude Code’s own tools gated by an allowlist string.

Bottom line for adoption: the coupling to Claude is small and in one place — about two files, plus the per-workflow tools allowlists and the sentinel-teaching prose. The engine would not change if you swapped agents. There is no abstract Agent interface — the seam is “one isolated function with an _argv_override test hook,” not a polymorphic boundary. To make it formally agent-neutral: pull out an Agent protocol invoke(prompt, tools, model, cwd, …) -> InvokeResult, and neutralize the tool-name vocabulary and the model-alias map. The structured-output contract is a sentinel line matched on free text — not JSON/tool-call/typed output — this is the weakest part and the first thing to replace if you adopt the model.

Appendix: the most important files

(all under /Users/sksizer2/Developer/DarkFactory/python/darkfactory/)

Agent call: utils/claude_code/_invoke.py, utils/claude_code/_interactive.py
Engine/dispatch: runner.py; engine/phase_state.py, engine/payloads.py
DAG: graph/_dag.py, graph/_execution.py, graph/_containment.py, graph/_assign.py
Workflow types: workflow/_core.py, workflow/_templates.py, workflow/_templates_builtin.py
Workflow defs: workflow/definitions/{prd,project}/*/workflow.py
Builtins: operations/_registry.py, operations/{ensure_worktree,set_status,commit,create_pr,analyze_transcript,project_builtins}.py
Rework: rework/{context,guard,prompt}.py
Plumbing: utils/git/, utils/github/, utils/shell.py, utils/_result.py
Config/state: config/, model/_persistence.py, event_log.py, timeouts.py, checks.py
CLI: cli/main.py, cli/run.py
Agent contract prose: workflow/definitions/prd/default/prompts/role.md, task.md

← Back to Notes