Orchestration Engine & Activities

Durable SQLite workflow engine, state transition manager, and stateless activity library.

dark_factory.orchestrator.engine.DurableEngine

Orchestrates durable runs, state transitions, and operator review gates.

Source code in dark_factory/orchestrator/engine.py

class DurableEngine:
    """Orchestrates durable runs, state transitions, and operator review gates."""

    def __init__(
        self,
        storage_dir: str | Path = ".factory",
        db_name: str = "factory.db",
    ) -> None:
        self.storage_dir = Path(storage_dir).resolve()
        self.storage_dir.mkdir(parents=True, exist_ok=True)
        self.db_path = self.storage_dir / db_name
        self.locker = EvidenceLocker(self.storage_dir)
        # Serializes the git-mutating part of review_run's approve path: unlike sandboxes (one isolated
        # worktree per run), activity_apply_patch mutates the host repo's actual working tree directly, so
        # concurrent approvals against the same repo would race on `checkout -b`/`apply`/`commit`.
        # Per-process only — see plan.md 10.11 for the cross-process limitation.
        self._review_lock = threading.Lock()
        self._init_db()

    @contextmanager
    def _get_connection(self) -> Generator[sqlite3.Connection, None, None]:
        conn = sqlite3.connect(str(self.db_path))
        conn.execute("PRAGMA journal_mode=WAL;")
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def _init_db(self) -> None:
        with self._get_connection() as conn:
            conn.executescript(
                """
                CREATE TABLE IF NOT EXISTS runs (
                    run_id TEXT PRIMARY KEY,
                    repo_path TEXT NOT NULL,
                    base_rev TEXT NOT NULL,
                    status TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    operator_notes TEXT
                );

                CREATE TABLE IF NOT EXISTS events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    run_id TEXT NOT NULL,
                    event_type TEXT NOT NULL,
                    payload TEXT,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY(run_id) REFERENCES runs(run_id)
                );
                """
            )

    def transition_status(
        self,
        run_id: str,
        new_status: RunStatus,
        payload: dict | None = None,
        operator_notes: str | None = None,
    ) -> None:
        """Atomically transition run status and record audit event."""
        now = datetime.now(UTC).isoformat()
        payload_json = json.dumps(payload or {})
        with self._get_connection() as conn:
            if operator_notes is not None:
                conn.execute(
                    """
                    UPDATE runs
                    SET status = ?, updated_at = ?, operator_notes = ?
                    WHERE run_id = ?;
                    """,
                    (new_status.value, now, operator_notes, run_id),
                )
            else:
                conn.execute(
                    """
                    UPDATE runs
                    SET status = ?, updated_at = ?
                    WHERE run_id = ?;
                    """,
                    (new_status.value, now, run_id),
                )
            conn.execute(
                """
                INSERT INTO events (run_id, event_type, payload, created_at)
                VALUES (?, ?, ?, ?);
                """,
                (run_id, new_status.value, payload_json, now),
            )

    def execute_run(
        self,
        spec: TaskSpec,
        harness: AgentHarness | None = None,
        run_id: str | None = None,
        status_callback: Callable[[RunStatus], None] | None = None,
    ) -> EvidenceManifest:
        """Execute a full autonomous dark factory run."""
        if run_id is None:
            now_str = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
            short_id = uuid.uuid4().hex[:6]
            run_id = f"run-{now_str}-{short_id}"

        if harness is None:
            harness = LocalCoderHarness(model=spec.model)

        now = datetime.now(UTC).isoformat()
        with self._get_connection() as conn:
            conn.execute(
                """
                INSERT INTO runs (run_id, repo_path, base_rev, status, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?);
                """,
                (run_id, str(spec.repo_path), spec.base_rev, RunStatus.PENDING.value, now, now),
            )

        manifest = EvidenceManifest.create(
            run_id=run_id,
            repo_path=str(spec.repo_path),
            base_rev=spec.base_rev,
            status=RunStatus.PENDING,
        )

        def update_state(st: RunStatus, pld=None):
            manifest.status = st
            self.transition_status(run_id, st, pld)
            if status_callback:
                status_callback(st)

        deadline = time.monotonic() + spec.timeout_minutes * 60
        sandbox = None
        executions: list[StepExecution] = []
        healing_attempts = 0
        telemetry: ModelTelemetry | None = None
        sampler = ResourceSampler()
        try:
            # 1. Create Sandbox
            update_state(RunStatus.CREATING_SANDBOX)
            with _trace(manifest, "sandbox_create"):
                sandbox = activity_create_sandbox(
                    spec=spec,
                    run_id=run_id,
                    base_dir=self.storage_dir / "sandboxes",
                )

            def finish_timed_out(timeout: RunTimeoutError) -> EvidenceManifest:
                manifest.status = RunStatus.TIMED_OUT
                manifest.operator_notes = f"Run exceeded its {spec.timeout_minutes:g} minute deadline."
                timeout_diff = self._best_effort_diff(sandbox)
                with _trace(manifest, "evidence_preserve"):
                    activity_preserve_evidence(
                        sandbox=sandbox,
                        locker=self.locker,
                        manifest=manifest,
                        executions=timeout.executions,
                        healing_attempts=timeout.healing_attempts,
                        telemetry=timeout.telemetry,
                        diff=timeout_diff,
                    )
                # The save above can't record its own duration; persist the now-complete phase_timings
                # (cheap: same patch content, no transcript arg so transcript.log is left untouched).
                self.locker.save_run(manifest, patch_content=timeout_diff)
                update_state(RunStatus.TIMED_OUT, {"reason": "deadline_exceeded"})
                return manifest

            # 2. Run Planning Session (if not skip_plan)
            plan = None
            if not spec.skip_plan:
                try:
                    with _trace(manifest, "task_planning"):
                        plan = activity_plan_task(
                            run_id=run_id,
                            spec=spec,
                            sandbox=sandbox,
                            manifest=manifest,
                            harness=harness,
                            deadline=deadline,
                        )
                except RunTimeoutError as timeout:
                    return finish_timed_out(timeout)

            # 3. Run Agent & Verification Loop
            update_state(RunStatus.AGENT_RUNNING)
            try:
                with _trace(manifest, "agent_and_verify"):
                    if not spec.skip_analysis:
                        sampler.start()
                    try:
                        passed, healing_attempts, executions, telemetry, repeated_failure_streak = (
                            activity_execute_task_and_verify(
                                sandbox=sandbox,
                                harness=harness,
                                spec=spec,
                                status_callback=update_state,
                                deadline=deadline,
                                plan=plan,
                            )
                        )
                    finally:
                        sampler.stop()
                manifest.repeated_failure_streak = repeated_failure_streak
            except RunTimeoutError as timeout:
                return finish_timed_out(timeout)

            # 3. Assess Result. Verification exit codes decide; a no-op is never a success.
            with _trace(manifest, "diff_extract"):
                diff = sandbox.get_diff()
            final_status = RunStatus.AWAITING_REVIEW if passed else RunStatus.FAILED
            final_payload: dict | None = None
            if passed and not diff.strip():
                final_status = RunStatus.FAILED
                final_payload = {"reason": "empty_patch"}
                manifest.operator_notes = "Verification passed but the agent produced no changes (empty patch)."
            elif not passed:
                if executions and executions[-1].step_id == "harness":
                    manifest.operator_notes = (
                        f"The agent harness failed after {healing_attempts} attempt(s): {executions[-1].stderr[:300]}"
                    )
                elif repeated_failure_streak >= 1:
                    manifest.operator_notes = (
                        f"Verification failed after exhausting {healing_attempts} healing attempt(s); the "
                        f"same failure repeated identically for the last {repeated_failure_streak + 1} "
                        "attempts (the model got stuck on one approach rather than trying something "
                        "different)."
                    )
                else:
                    manifest.operator_notes = (
                        f"Verification failed after exhausting {healing_attempts} healing attempt(s); each "
                        "attempt failed differently."
                    )
            elif spec.mutate_adversarial:
                # 3.1. Active Adversarial Test Mutation Gate
                try:
                    with _trace(manifest, "adversarial_mutation"):
                        mut_passed, mut_attempts, mut_execs, mut_telem = activity_adversarial_mutation(
                            sandbox=sandbox,
                            harness=harness,
                            spec=spec,
                            manifest=manifest,
                            diff=diff,
                            status_callback=update_state,
                            deadline=deadline,
                            healing_attempts_so_far=healing_attempts,
                        )
                except RunTimeoutError as timeout:
                    timeout.executions = executions + timeout.executions
                    timeout.healing_attempts += healing_attempts
                    timeout.telemetry = timeout.telemetry or telemetry
                    return finish_timed_out(timeout)
                executions.extend(mut_execs)
                healing_attempts += mut_attempts
                if mut_telem and telemetry:
                    telemetry.prompt_tokens += mut_telem.prompt_tokens
                    telemetry.completion_tokens += mut_telem.completion_tokens
                    telemetry.total_tokens += mut_telem.total_tokens
                    telemetry.duration_sec = round(telemetry.duration_sec + mut_telem.duration_sec, 3)
                elif mut_telem:
                    telemetry = mut_telem

                if not mut_passed:
                    passed = False
                    final_status = RunStatus.FAILED
                    reverify_failed = any(ex.step_id.endswith(":reverify") and not ex.passed for ex in executions)
                    manifest.operator_notes = (
                        "Verification failed: baseline gates failed after the adversarial probe ran "
                        "(post-probe re-verification); the probe may have altered the sources."
                        if reverify_failed
                        else f"Verification failed: adversarial probe test failed after {healing_attempts} healing attempt(s)."
                    )
                else:
                    # Re-extract diff in case healing modified files
                    with _trace(manifest, "diff_extract"):
                        diff = sandbox.get_diff()
                    if not spec.skip_adversarial:
                        # 3.2. Adversarial Red-Team Audit Gate (on verified final diff)
                        gate_summary_lines = [
                            f"[{ex.step_id}] {'PASSED' if ex.passed else 'FAILED'}" for ex in executions
                        ]
                        try:
                            with _trace(manifest, "adversarial_audit"):
                                activity_adversarial_audit(
                                    spec=spec,
                                    diff=diff,
                                    manifest=manifest,
                                    harness=harness,
                                    gate_summary="\n".join(gate_summary_lines),
                                    deadline=deadline,
                                )
                        except RunTimeoutError as timeout:
                            timeout.executions = executions + timeout.executions
                            timeout.healing_attempts += healing_attempts
                            timeout.telemetry = timeout.telemetry or telemetry
                            return finish_timed_out(timeout)
            elif not spec.skip_adversarial:
                # 3.1. Adversarial Red-Team Audit Gate (only on verified, non-empty patches)
                gate_summary_lines = [f"[{ex.step_id}] {'PASSED' if ex.passed else 'FAILED'}" for ex in executions]
                try:
                    with _trace(manifest, "adversarial_audit"):
                        activity_adversarial_audit(
                            spec=spec,
                            diff=diff,
                            manifest=manifest,
                            harness=harness,
                            gate_summary="\n".join(gate_summary_lines),
                            deadline=deadline,
                        )
                except RunTimeoutError as timeout:
                    timeout.executions = executions + timeout.executions
                    timeout.healing_attempts += healing_attempts
                    timeout.telemetry = timeout.telemetry or telemetry
                    return finish_timed_out(timeout)

            # 3.8. Patch-scope gate: a verified patch that touches paths outside the task's scope is not reviewable.
            if final_status == RunStatus.AWAITING_REVIEW:
                out_of_scope = scope_violations(diff, spec.target_files)
                if out_of_scope:
                    listed = ", ".join(out_of_scope[:10]) + (
                        f" (+{len(out_of_scope) - 10} more)" if len(out_of_scope) > 10 else ""
                    )
                    if spec.allow_out_of_scope:
                        manifest.operator_notes = f"Out-of-scope patch check bypassed (--allow-out-of-scope): {listed}."
                    else:
                        final_status = RunStatus.FAILED
                        final_payload = {"reason": "out_of_scope_patch", "paths": out_of_scope}
                        manifest.operator_notes = (
                            f"Verification passed but the patch touches out-of-scope paths: {listed}. "
                            "The diff is preserved; rerun with --allow-out-of-scope to accept it."
                        )

            # 3.9. Advisory performance & quality analysis (verified, non-empty patches only; never alters status).
            if final_status == RunStatus.AWAITING_REVIEW and not spec.skip_analysis:
                with _trace(manifest, "analysis"):
                    # Real run values; the manifest only receives them in activity_preserve_evidence below.
                    manifest.model_telemetry = telemetry
                    manifest.healing_attempts = healing_attempts
                    try:
                        manifest.analysis_report = analyze(
                            manifest, diff, spec, sampler.usage(), sources=collect_sources(sandbox, diff)
                        )
                    except Exception as exc:
                        manifest.analysis_report = AnalysisReport(
                            summary="Analysis unavailable.",
                            findings=[AnalysisFinding("INFO", "quality", "Analysis failed", str(exc))],
                        )

            # 4. Preserve evidence BEFORE the durable status transition, so a crash can never leave a
            #    review-pending run without its evidence (recover() reconciles the opposite window).
            manifest.status = final_status
            with _trace(manifest, "evidence_preserve"):
                activity_preserve_evidence(
                    sandbox=sandbox,
                    locker=self.locker,
                    manifest=manifest,
                    executions=executions,
                    healing_attempts=healing_attempts,
                    telemetry=telemetry,
                    diff=diff,
                )
            # The save above can't record its own duration; persist the now-complete phase_timings
            # (cheap: same patch content, no transcript arg so transcript.log is left untouched).
            self.locker.save_run(manifest, patch_content=diff)
            update_state(final_status, final_payload)

            return manifest

        except (KeyboardInterrupt, SystemExit) as interrupt:
            manifest.status = RunStatus.CANCELLED
            manifest.operator_notes = f"Run cancelled ({type(interrupt).__name__})."
            try:
                if sandbox is not None:
                    # Same evidence shape as any other terminal state: preserve whatever diff the
                    # harness had produced before the interrupt, not just the manifest/transcript.
                    cancel_diff = self._best_effort_diff(sandbox)
                    with _trace(manifest, "evidence_preserve"):
                        activity_preserve_evidence(
                            sandbox=sandbox,
                            locker=self.locker,
                            manifest=manifest,
                            executions=executions,
                            healing_attempts=healing_attempts,
                            telemetry=telemetry,
                            diff=cancel_diff,
                        )
                    # The save above can't record its own duration; persist the now-complete
                    # phase_timings (cheap: same patch content, no transcript arg).
                    self.locker.save_run(manifest, patch_content=cancel_diff)
                else:
                    self.locker.save_run(manifest, transcript=f"CANCELLED: {type(interrupt).__name__}")
                self.transition_status(run_id, RunStatus.CANCELLED, {"reason": type(interrupt).__name__})
            except Exception:
                pass  # never mask the interrupt; recover() will reconcile
            raise

        except Exception as e:
            import traceback

            tb = traceback.format_exc()
            update_state(RunStatus.FAILED, {"error": str(e), "traceback": tb})
            manifest.status = RunStatus.FAILED
            manifest.operator_notes = f"Run encountered unhandled error: {e}"
            self.locker.save_run(manifest, transcript=f"ERROR: {e}\n\nTRACEBACK:\n{tb}")
            raise
        finally:
            if sandbox is not None:
                activity_cleanup_sandbox(sandbox)

    @staticmethod
    def _best_effort_diff(sandbox) -> str:
        try:
            return sandbox.get_diff()
        except Exception:
            return ""

    def review_run(
        self,
        run_id: str,
        approve: bool,
        target_branch: str | None = None,
        note: str | None = None,
    ) -> EvidenceManifest:
        """Handle human-in-the-loop review decision."""
        manifest = self.locker.load_manifest(run_id)
        if manifest.status != RunStatus.AWAITING_REVIEW:
            raise WorkflowStateError(
                f"Cannot review run '{run_id}' with status '{manifest.status.value}' (expected AWAITING_REVIEW)."
            )

        now = datetime.now(UTC).isoformat()

        if approve:
            patch = self.locker.load_patch(run_id)
            if manifest.patch_sha256:
                actual_sha = hashlib.sha256(patch.encode("utf-8")).hexdigest()
                if actual_sha != manifest.patch_sha256:
                    raise WorkflowStateError(
                        f"Patch integrity check failed for run '{run_id}'! "
                        f"Expected SHA256 {manifest.patch_sha256}, got {actual_sha}."
                    )

            # Serialize the actual host-repo mutation: `git checkout -b` / `apply` / `commit` operate on
            # one shared working tree, so two approvals racing here would corrupt each other.
            with self._review_lock, _trace(manifest, "patch_apply"):
                resulting_rev = activity_apply_patch(
                    repo_path=manifest.repo_path,
                    patch_content=patch,
                    target_branch=target_branch,
                    commit_msg=f"feat(dark-factory): applied verified patch from {run_id}",
                )
            manifest.status = RunStatus.APPROVED
            manifest.resulting_rev = resulting_rev
            manifest.operator_notes = note or "Approved by operator."
            manifest.completed_at = now
            self.transition_status(
                run_id,
                RunStatus.APPROVED,
                payload={"resulting_rev": resulting_rev},
                operator_notes=manifest.operator_notes,
            )
        else:
            manifest.status = RunStatus.REJECTED
            manifest.operator_notes = note or "Rejected by operator."
            manifest.completed_at = now
            self.transition_status(
                run_id,
                RunStatus.REJECTED,
                payload={"note": manifest.operator_notes},
                operator_notes=manifest.operator_notes,
            )

        self.locker.save_run(manifest, patch_content=self.locker.load_patch(run_id))
        return manifest

    def list_runs(self) -> list[dict]:
        """List all runs recorded in SQLite."""
        with self._get_connection() as conn:
            rows = conn.execute(
                "SELECT run_id, repo_path, base_rev, status, created_at, updated_at, operator_notes FROM runs ORDER BY created_at DESC;"
            ).fetchall()
            return [dict(row) for row in rows]

    def get_run(self, run_id: str) -> EvidenceManifest:
        """Fetch full evidence manifest for a run."""
        return self.locker.load_manifest(run_id)

    def _reconcile_orphan(self, run_id: str) -> None:
        """Settle a run whose process died: adopt its evidence's outcome if complete, else mark FAILED.

        Evidence is written before the DB transition, and manifest.json is written last and atomically, so a
        readable manifest with a matching patch digest proves the run really reached that outcome.
        """
        try:
            manifest = self.locker.load_manifest(run_id)
            if manifest.status == RunStatus.AWAITING_REVIEW:
                patch = self.locker.load_patch(run_id)
                digest = hashlib.sha256(patch.encode("utf-8")).hexdigest()
                if manifest.patch_sha256 and digest == manifest.patch_sha256:
                    self.transition_status(
                        run_id,
                        RunStatus.AWAITING_REVIEW,
                        {"reason": "Engine restart recovery: evidence complete"},
                    )
                    return
            elif manifest.status in (RunStatus.FAILED, RunStatus.TIMED_OUT, RunStatus.CANCELLED):
                self.transition_status(run_id, manifest.status, {"reason": "Engine restart recovery"})
                return
        except Exception:
            pass  # no readable evidence: fall through to FAILED
        self.transition_status(run_id, RunStatus.FAILED, {"reason": "Engine restart recovery"})

    def _reconcile_review_completeness(self, run_id: str) -> bool:
        """Detect a crash between `transition_status(APPROVED/REJECTED)` and the final `locker.save_run()`
        in `review_run()` (DB says the review concluded, but the manifest on disk never caught up).

        For APPROVED, the patch may already have been committed to the target branch by
        `activity_apply_patch` before the crash, so we must not re-derive or re-apply it; the only safe
        move is to surface the inconsistency as `FAILED` with an explanatory note so the operator checks
        the repository by hand. Returns True if an inconsistency was found and flagged.
        """
        with self._get_connection() as conn:
            row = conn.execute("SELECT status FROM runs WHERE run_id = ?;", (run_id,)).fetchone()
        if row is None:
            return False
        db_status = row["status"]

        try:
            manifest = self.locker.load_manifest(run_id)
        except Exception:
            manifest = None

        if db_status == RunStatus.APPROVED.value:
            complete = manifest is not None and manifest.status == RunStatus.APPROVED and manifest.resulting_rev
        elif db_status == RunStatus.REJECTED.value:
            complete = manifest is not None and manifest.status == RunStatus.REJECTED and manifest.completed_at
        else:
            return False
        if complete:
            return False

        self.transition_status(
            run_id,
            RunStatus.FAILED,
            {
                "reason": (
                    f"Engine restart recovery: {db_status} transition did not complete "
                    "(evidence write crashed); verify repository state manually before retrying."
                )
            },
        )
        return True

    def recover(self) -> list[str]:
        """Recover orphaned runs, clean their sandboxes, and prune git worktrees."""
        recovered = []
        affected_repos: set[str] = set()
        sandboxes_dir = self.storage_dir / "sandboxes"

        with self._get_connection() as conn:
            placeholders = ",".join("?" for _ in _NON_ORPHAN_STATUSES)
            rows = conn.execute(
                f"SELECT run_id, repo_path FROM runs WHERE status NOT IN ({placeholders});",
                tuple(_NON_ORPHAN_STATUSES),
            ).fetchall()
            for row in rows:
                rid = row["run_id"]
                repo_path = row["repo_path"]
                if repo_path:
                    affected_repos.add(repo_path)
                self._reconcile_orphan(rid)
                recovered.append(rid)

                # Prune dead sandbox folder for this run
                run_sandbox = sandboxes_dir / rid
                if run_sandbox.exists():
                    import shutil

                    shutil.rmtree(run_sandbox, ignore_errors=True)

            # Separately: catch runs whose DB status already reads APPROVED/REJECTED but whose evidence
            # never caught up (crash between transition_status and the final locker.save_run in review_run).
            review_rows = conn.execute(
                "SELECT run_id FROM runs WHERE status IN (?, ?);",
                (RunStatus.APPROVED.value, RunStatus.REJECTED.value),
            ).fetchall()

        for row in review_rows:
            rid = row["run_id"]
            if self._reconcile_review_completeness(rid):
                recovered.append(rid)

        # Prune git worktree metadata in affected repos
        for repo_str in affected_repos:
            repo_p = Path(repo_str)
            if repo_p.exists():
                import subprocess

                subprocess.run(["git", "worktree", "prune"], cwd=repo_p, check=False, capture_output=True)

        return recovered

Methods:

execute_run(spec, harness=None, run_id=None, status_callback=None)

Execute a full autonomous dark factory run.

Source code in dark_factory/orchestrator/engine.py

def execute_run(
    self,
    spec: TaskSpec,
    harness: AgentHarness | None = None,
    run_id: str | None = None,
    status_callback: Callable[[RunStatus], None] | None = None,
) -> EvidenceManifest:
    """Execute a full autonomous dark factory run."""
    if run_id is None:
        now_str = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
        short_id = uuid.uuid4().hex[:6]
        run_id = f"run-{now_str}-{short_id}"

    if harness is None:
        harness = LocalCoderHarness(model=spec.model)

    now = datetime.now(UTC).isoformat()
    with self._get_connection() as conn:
        conn.execute(
            """
            INSERT INTO runs (run_id, repo_path, base_rev, status, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?);
            """,
            (run_id, str(spec.repo_path), spec.base_rev, RunStatus.PENDING.value, now, now),
        )

    manifest = EvidenceManifest.create(
        run_id=run_id,
        repo_path=str(spec.repo_path),
        base_rev=spec.base_rev,
        status=RunStatus.PENDING,
    )

    def update_state(st: RunStatus, pld=None):
        manifest.status = st
        self.transition_status(run_id, st, pld)
        if status_callback:
            status_callback(st)

    deadline = time.monotonic() + spec.timeout_minutes * 60
    sandbox = None
    executions: list[StepExecution] = []
    healing_attempts = 0
    telemetry: ModelTelemetry | None = None
    sampler = ResourceSampler()
    try:
        # 1. Create Sandbox
        update_state(RunStatus.CREATING_SANDBOX)
        with _trace(manifest, "sandbox_create"):
            sandbox = activity_create_sandbox(
                spec=spec,
                run_id=run_id,
                base_dir=self.storage_dir / "sandboxes",
            )

        def finish_timed_out(timeout: RunTimeoutError) -> EvidenceManifest:
            manifest.status = RunStatus.TIMED_OUT
            manifest.operator_notes = f"Run exceeded its {spec.timeout_minutes:g} minute deadline."
            timeout_diff = self._best_effort_diff(sandbox)
            with _trace(manifest, "evidence_preserve"):
                activity_preserve_evidence(
                    sandbox=sandbox,
                    locker=self.locker,
                    manifest=manifest,
                    executions=timeout.executions,
                    healing_attempts=timeout.healing_attempts,
                    telemetry=timeout.telemetry,
                    diff=timeout_diff,
                )
            # The save above can't record its own duration; persist the now-complete phase_timings
            # (cheap: same patch content, no transcript arg so transcript.log is left untouched).
            self.locker.save_run(manifest, patch_content=timeout_diff)
            update_state(RunStatus.TIMED_OUT, {"reason": "deadline_exceeded"})
            return manifest

        # 2. Run Planning Session (if not skip_plan)
        plan = None
        if not spec.skip_plan:
            try:
                with _trace(manifest, "task_planning"):
                    plan = activity_plan_task(
                        run_id=run_id,
                        spec=spec,
                        sandbox=sandbox,
                        manifest=manifest,
                        harness=harness,
                        deadline=deadline,
                    )
            except RunTimeoutError as timeout:
                return finish_timed_out(timeout)

        # 3. Run Agent & Verification Loop
        update_state(RunStatus.AGENT_RUNNING)
        try:
            with _trace(manifest, "agent_and_verify"):
                if not spec.skip_analysis:
                    sampler.start()
                try:
                    passed, healing_attempts, executions, telemetry, repeated_failure_streak = (
                        activity_execute_task_and_verify(
                            sandbox=sandbox,
                            harness=harness,
                            spec=spec,
                            status_callback=update_state,
                            deadline=deadline,
                            plan=plan,
                        )
                    )
                finally:
                    sampler.stop()
            manifest.repeated_failure_streak = repeated_failure_streak
        except RunTimeoutError as timeout:
            return finish_timed_out(timeout)

        # 3. Assess Result. Verification exit codes decide; a no-op is never a success.
        with _trace(manifest, "diff_extract"):
            diff = sandbox.get_diff()
        final_status = RunStatus.AWAITING_REVIEW if passed else RunStatus.FAILED
        final_payload: dict | None = None
        if passed and not diff.strip():
            final_status = RunStatus.FAILED
            final_payload = {"reason": "empty_patch"}
            manifest.operator_notes = "Verification passed but the agent produced no changes (empty patch)."
        elif not passed:
            if executions and executions[-1].step_id == "harness":
                manifest.operator_notes = (
                    f"The agent harness failed after {healing_attempts} attempt(s): {executions[-1].stderr[:300]}"
                )
            elif repeated_failure_streak >= 1:
                manifest.operator_notes = (
                    f"Verification failed after exhausting {healing_attempts} healing attempt(s); the "
                    f"same failure repeated identically for the last {repeated_failure_streak + 1} "
                    "attempts (the model got stuck on one approach rather than trying something "
                    "different)."
                )
            else:
                manifest.operator_notes = (
                    f"Verification failed after exhausting {healing_attempts} healing attempt(s); each "
                    "attempt failed differently."
                )
        elif spec.mutate_adversarial:
            # 3.1. Active Adversarial Test Mutation Gate
            try:
                with _trace(manifest, "adversarial_mutation"):
                    mut_passed, mut_attempts, mut_execs, mut_telem = activity_adversarial_mutation(
                        sandbox=sandbox,
                        harness=harness,
                        spec=spec,
                        manifest=manifest,
                        diff=diff,
                        status_callback=update_state,
                        deadline=deadline,
                        healing_attempts_so_far=healing_attempts,
                    )
            except RunTimeoutError as timeout:
                timeout.executions = executions + timeout.executions
                timeout.healing_attempts += healing_attempts
                timeout.telemetry = timeout.telemetry or telemetry
                return finish_timed_out(timeout)
            executions.extend(mut_execs)
            healing_attempts += mut_attempts
            if mut_telem and telemetry:
                telemetry.prompt_tokens += mut_telem.prompt_tokens
                telemetry.completion_tokens += mut_telem.completion_tokens
                telemetry.total_tokens += mut_telem.total_tokens
                telemetry.duration_sec = round(telemetry.duration_sec + mut_telem.duration_sec, 3)
            elif mut_telem:
                telemetry = mut_telem

            if not mut_passed:
                passed = False
                final_status = RunStatus.FAILED
                reverify_failed = any(ex.step_id.endswith(":reverify") and not ex.passed for ex in executions)
                manifest.operator_notes = (
                    "Verification failed: baseline gates failed after the adversarial probe ran "
                    "(post-probe re-verification); the probe may have altered the sources."
                    if reverify_failed
                    else f"Verification failed: adversarial probe test failed after {healing_attempts} healing attempt(s)."
                )
            else:
                # Re-extract diff in case healing modified files
                with _trace(manifest, "diff_extract"):
                    diff = sandbox.get_diff()
                if not spec.skip_adversarial:
                    # 3.2. Adversarial Red-Team Audit Gate (on verified final diff)
                    gate_summary_lines = [
                        f"[{ex.step_id}] {'PASSED' if ex.passed else 'FAILED'}" for ex in executions
                    ]
                    try:
                        with _trace(manifest, "adversarial_audit"):
                            activity_adversarial_audit(
                                spec=spec,
                                diff=diff,
                                manifest=manifest,
                                harness=harness,
                                gate_summary="\n".join(gate_summary_lines),
                                deadline=deadline,
                            )
                    except RunTimeoutError as timeout:
                        timeout.executions = executions + timeout.executions
                        timeout.healing_attempts += healing_attempts
                        timeout.telemetry = timeout.telemetry or telemetry
                        return finish_timed_out(timeout)
        elif not spec.skip_adversarial:
            # 3.1. Adversarial Red-Team Audit Gate (only on verified, non-empty patches)
            gate_summary_lines = [f"[{ex.step_id}] {'PASSED' if ex.passed else 'FAILED'}" for ex in executions]
            try:
                with _trace(manifest, "adversarial_audit"):
                    activity_adversarial_audit(
                        spec=spec,
                        diff=diff,
                        manifest=manifest,
                        harness=harness,
                        gate_summary="\n".join(gate_summary_lines),
                        deadline=deadline,
                    )
            except RunTimeoutError as timeout:
                timeout.executions = executions + timeout.executions
                timeout.healing_attempts += healing_attempts
                timeout.telemetry = timeout.telemetry or telemetry
                return finish_timed_out(timeout)

        # 3.8. Patch-scope gate: a verified patch that touches paths outside the task's scope is not reviewable.
        if final_status == RunStatus.AWAITING_REVIEW:
            out_of_scope = scope_violations(diff, spec.target_files)
            if out_of_scope:
                listed = ", ".join(out_of_scope[:10]) + (
                    f" (+{len(out_of_scope) - 10} more)" if len(out_of_scope) > 10 else ""
                )
                if spec.allow_out_of_scope:
                    manifest.operator_notes = f"Out-of-scope patch check bypassed (--allow-out-of-scope): {listed}."
                else:
                    final_status = RunStatus.FAILED
                    final_payload = {"reason": "out_of_scope_patch", "paths": out_of_scope}
                    manifest.operator_notes = (
                        f"Verification passed but the patch touches out-of-scope paths: {listed}. "
                        "The diff is preserved; rerun with --allow-out-of-scope to accept it."
                    )

        # 3.9. Advisory performance & quality analysis (verified, non-empty patches only; never alters status).
        if final_status == RunStatus.AWAITING_REVIEW and not spec.skip_analysis:
            with _trace(manifest, "analysis"):
                # Real run values; the manifest only receives them in activity_preserve_evidence below.
                manifest.model_telemetry = telemetry
                manifest.healing_attempts = healing_attempts
                try:
                    manifest.analysis_report = analyze(
                        manifest, diff, spec, sampler.usage(), sources=collect_sources(sandbox, diff)
                    )
                except Exception as exc:
                    manifest.analysis_report = AnalysisReport(
                        summary="Analysis unavailable.",
                        findings=[AnalysisFinding("INFO", "quality", "Analysis failed", str(exc))],
                    )

        # 4. Preserve evidence BEFORE the durable status transition, so a crash can never leave a
        #    review-pending run without its evidence (recover() reconciles the opposite window).
        manifest.status = final_status
        with _trace(manifest, "evidence_preserve"):
            activity_preserve_evidence(
                sandbox=sandbox,
                locker=self.locker,
                manifest=manifest,
                executions=executions,
                healing_attempts=healing_attempts,
                telemetry=telemetry,
                diff=diff,
            )
        # The save above can't record its own duration; persist the now-complete phase_timings
        # (cheap: same patch content, no transcript arg so transcript.log is left untouched).
        self.locker.save_run(manifest, patch_content=diff)
        update_state(final_status, final_payload)

        return manifest

    except (KeyboardInterrupt, SystemExit) as interrupt:
        manifest.status = RunStatus.CANCELLED
        manifest.operator_notes = f"Run cancelled ({type(interrupt).__name__})."
        try:
            if sandbox is not None:
                # Same evidence shape as any other terminal state: preserve whatever diff the
                # harness had produced before the interrupt, not just the manifest/transcript.
                cancel_diff = self._best_effort_diff(sandbox)
                with _trace(manifest, "evidence_preserve"):
                    activity_preserve_evidence(
                        sandbox=sandbox,
                        locker=self.locker,
                        manifest=manifest,
                        executions=executions,
                        healing_attempts=healing_attempts,
                        telemetry=telemetry,
                        diff=cancel_diff,
                    )
                # The save above can't record its own duration; persist the now-complete
                # phase_timings (cheap: same patch content, no transcript arg).
                self.locker.save_run(manifest, patch_content=cancel_diff)
            else:
                self.locker.save_run(manifest, transcript=f"CANCELLED: {type(interrupt).__name__}")
            self.transition_status(run_id, RunStatus.CANCELLED, {"reason": type(interrupt).__name__})
        except Exception:
            pass  # never mask the interrupt; recover() will reconcile
        raise

    except Exception as e:
        import traceback

        tb = traceback.format_exc()
        update_state(RunStatus.FAILED, {"error": str(e), "traceback": tb})
        manifest.status = RunStatus.FAILED
        manifest.operator_notes = f"Run encountered unhandled error: {e}"
        self.locker.save_run(manifest, transcript=f"ERROR: {e}\n\nTRACEBACK:\n{tb}")
        raise
    finally:
        if sandbox is not None:
            activity_cleanup_sandbox(sandbox)

get_run(run_id)

Fetch full evidence manifest for a run.

Source code in dark_factory/orchestrator/engine.py

def get_run(self, run_id: str) -> EvidenceManifest:
    """Fetch full evidence manifest for a run."""
    return self.locker.load_manifest(run_id)

list_runs()

List all runs recorded in SQLite.

Source code in dark_factory/orchestrator/engine.py

def list_runs(self) -> list[dict]:
    """List all runs recorded in SQLite."""
    with self._get_connection() as conn:
        rows = conn.execute(
            "SELECT run_id, repo_path, base_rev, status, created_at, updated_at, operator_notes FROM runs ORDER BY created_at DESC;"
        ).fetchall()
        return [dict(row) for row in rows]

recover()

Recover orphaned runs, clean their sandboxes, and prune git worktrees.

Source code in dark_factory/orchestrator/engine.py

def recover(self) -> list[str]:
    """Recover orphaned runs, clean their sandboxes, and prune git worktrees."""
    recovered = []
    affected_repos: set[str] = set()
    sandboxes_dir = self.storage_dir / "sandboxes"

    with self._get_connection() as conn:
        placeholders = ",".join("?" for _ in _NON_ORPHAN_STATUSES)
        rows = conn.execute(
            f"SELECT run_id, repo_path FROM runs WHERE status NOT IN ({placeholders});",
            tuple(_NON_ORPHAN_STATUSES),
        ).fetchall()
        for row in rows:
            rid = row["run_id"]
            repo_path = row["repo_path"]
            if repo_path:
                affected_repos.add(repo_path)
            self._reconcile_orphan(rid)
            recovered.append(rid)

            # Prune dead sandbox folder for this run
            run_sandbox = sandboxes_dir / rid
            if run_sandbox.exists():
                import shutil

                shutil.rmtree(run_sandbox, ignore_errors=True)

        # Separately: catch runs whose DB status already reads APPROVED/REJECTED but whose evidence
        # never caught up (crash between transition_status and the final locker.save_run in review_run).
        review_rows = conn.execute(
            "SELECT run_id FROM runs WHERE status IN (?, ?);",
            (RunStatus.APPROVED.value, RunStatus.REJECTED.value),
        ).fetchall()

    for row in review_rows:
        rid = row["run_id"]
        if self._reconcile_review_completeness(rid):
            recovered.append(rid)

    # Prune git worktree metadata in affected repos
    for repo_str in affected_repos:
        repo_p = Path(repo_str)
        if repo_p.exists():
            import subprocess

            subprocess.run(["git", "worktree", "prune"], cwd=repo_p, check=False, capture_output=True)

    return recovered

review_run(run_id, approve, target_branch=None, note=None)

Handle human-in-the-loop review decision.

Source code in dark_factory/orchestrator/engine.py

def review_run(
    self,
    run_id: str,
    approve: bool,
    target_branch: str | None = None,
    note: str | None = None,
) -> EvidenceManifest:
    """Handle human-in-the-loop review decision."""
    manifest = self.locker.load_manifest(run_id)
    if manifest.status != RunStatus.AWAITING_REVIEW:
        raise WorkflowStateError(
            f"Cannot review run '{run_id}' with status '{manifest.status.value}' (expected AWAITING_REVIEW)."
        )

    now = datetime.now(UTC).isoformat()

    if approve:
        patch = self.locker.load_patch(run_id)
        if manifest.patch_sha256:
            actual_sha = hashlib.sha256(patch.encode("utf-8")).hexdigest()
            if actual_sha != manifest.patch_sha256:
                raise WorkflowStateError(
                    f"Patch integrity check failed for run '{run_id}'! "
                    f"Expected SHA256 {manifest.patch_sha256}, got {actual_sha}."
                )

        # Serialize the actual host-repo mutation: `git checkout -b` / `apply` / `commit` operate on
        # one shared working tree, so two approvals racing here would corrupt each other.
        with self._review_lock, _trace(manifest, "patch_apply"):
            resulting_rev = activity_apply_patch(
                repo_path=manifest.repo_path,
                patch_content=patch,
                target_branch=target_branch,
                commit_msg=f"feat(dark-factory): applied verified patch from {run_id}",
            )
        manifest.status = RunStatus.APPROVED
        manifest.resulting_rev = resulting_rev
        manifest.operator_notes = note or "Approved by operator."
        manifest.completed_at = now
        self.transition_status(
            run_id,
            RunStatus.APPROVED,
            payload={"resulting_rev": resulting_rev},
            operator_notes=manifest.operator_notes,
        )
    else:
        manifest.status = RunStatus.REJECTED
        manifest.operator_notes = note or "Rejected by operator."
        manifest.completed_at = now
        self.transition_status(
            run_id,
            RunStatus.REJECTED,
            payload={"note": manifest.operator_notes},
            operator_notes=manifest.operator_notes,
        )

    self.locker.save_run(manifest, patch_content=self.locker.load_patch(run_id))
    return manifest

transition_status(run_id, new_status, payload=None, operator_notes=None)

Atomically transition run status and record audit event.

Source code in dark_factory/orchestrator/engine.py

def transition_status(
    self,
    run_id: str,
    new_status: RunStatus,
    payload: dict | None = None,
    operator_notes: str | None = None,
) -> None:
    """Atomically transition run status and record audit event."""
    now = datetime.now(UTC).isoformat()
    payload_json = json.dumps(payload or {})
    with self._get_connection() as conn:
        if operator_notes is not None:
            conn.execute(
                """
                UPDATE runs
                SET status = ?, updated_at = ?, operator_notes = ?
                WHERE run_id = ?;
                """,
                (new_status.value, now, operator_notes, run_id),
            )
        else:
            conn.execute(
                """
                UPDATE runs
                SET status = ?, updated_at = ?
                WHERE run_id = ?;
                """,
                (new_status.value, now, run_id),
            )
        conn.execute(
            """
            INSERT INTO events (run_id, event_type, payload, created_at)
            VALUES (?, ?, ?, ?);
            """,
            (run_id, new_status.value, payload_json, now),
        )

dark_factory.orchestrator.activities

Stateless workflow activities for Sovereign Dark Factory (Option C Architecture).

These functions encapsulate the core steps of a factory run and can be executed directly by the embedded SQLite engine. Option C is a forward-looking seam; no Temporal worker is registered in this release.

Classes

Functions:

activity_adversarial_audit(spec, diff, manifest, harness=None, auditor=None, gate_summary='', deadline=None)

Execute adversarial red-team audit on the generated diff before human review.

Source code in dark_factory/orchestrator/activities.py

def activity_adversarial_audit(
    spec: TaskSpec,
    diff: str,
    manifest: EvidenceManifest,
    harness: AgentHarness | None = None,
    auditor: AdversarialAuditor | None = None,
    gate_summary: str = "",
    deadline: float | None = None,
) -> AdversarialReport | None:
    """Execute adversarial red-team audit on the generated diff before human review."""
    if spec.skip_adversarial:
        return None

    if auditor is None:
        from dark_factory.harness.local_coder import LocalCoderHarness

        if harness is not None and not isinstance(harness, LocalCoderHarness) and not hasattr(harness, "_call_model"):
            report = AdversarialReport(passed=True, summary="Deterministic audit passed.", findings=[])
            manifest.adversarial_report = report
            return report
        auditor = AdversarialAuditor(harness=harness)  # type: ignore[arg-type]

    with _clamped_harness(auditor, deadline):
        report = auditor.audit_patch(
            task_prompt=spec.task_prompt,
            patch=diff,
            gate_summary=gate_summary,
        )
    manifest.adversarial_report = report
    return report

activity_adversarial_mutation(sandbox, harness, spec, manifest, diff, status_callback=None, deadline=None, mutator=None, healer=None, healing_attempts_so_far=0)

Synthesize active hostile unit tests to probe the patch and heal if probe fails.

Source code in dark_factory/orchestrator/activities.py

def activity_adversarial_mutation(
    sandbox: Sandbox,
    harness: AgentHarness,
    spec: TaskSpec,
    manifest: EvidenceManifest,
    diff: str,
    status_callback: Callable[[RunStatus], None] | None = None,
    deadline: float | None = None,
    mutator: AdversarialMutator | None = None,
    healer: SelfHealingLoop | None = None,
    healing_attempts_so_far: int = 0,
) -> tuple[bool, int, list[StepExecution], ModelTelemetry | None]:
    """Synthesize active hostile unit tests to probe the patch and heal if probe fails."""
    if not spec.mutate_adversarial or not diff.strip():
        return True, 0, [], None

    from dark_factory.harness.local_coder import LocalCoderHarness
    from dark_factory.verification.adversarial_mutator import AdversarialMutator

    if mutator is None:
        mut_harness = harness if isinstance(harness, LocalCoderHarness) else LocalCoderHarness(model=spec.model)
        mutator = AdversarialMutator(harness=mut_harness)

    # Lightweight context
    context = ""
    try:
        res = sandbox.execute(["git", "ls-files"])
        if res.passed and res.stdout.strip():
            context = f"Files in repository:\n{res.stdout.strip()[:1000]}"
    except Exception:
        pass

    with _clamped_harness(mutator, deadline):
        probe_code = mutator.generate_probe(task_prompt=spec.task_prompt, patch=diff, context=context)
    if not probe_code:
        # Graceful fallback: syntax error or generation failure in probe
        return True, 0, [], None

    manifest.adversarial_test_code = probe_code
    probe_path = "test_adversarial_probe.py"
    sandbox.write_file(probe_path, probe_code.encode("utf-8"))

    # Probe step
    probe_step = VerificationStep(
        id="adversarial_probe",
        argv=[sys.executable, "-m", "pytest", probe_path],
        timeout_sec=60,
    )
    combined_steps = list(spec.verification_steps) + [probe_step]

    class AdversarialVerificationRunner(VerificationRunner):
        def run(self, sbox: Sandbox, deadline: float | None = None) -> VerificationOutcome:
            sbox.write_file(probe_path, probe_code.encode("utf-8"))
            return super().run(sbox, deadline=deadline)

    runner = AdversarialVerificationRunner(
        steps=combined_steps,
        allow_no_verify=spec.allow_no_verify,
        protected_paths=spec.protected_paths,
        allow_gate_edits=spec.allow_gate_edits,
        base_rev=spec.base_rev,
    )

    def cleanup_probe():
        try:
            sandbox.restore_paths([probe_path])
        except Exception:
            pass
        if hasattr(sandbox, "path"):
            try:
                (Path(sandbox.path) / probe_path).unlink(missing_ok=True)
            except Exception:
                pass

    def reverify_baseline() -> tuple[bool, list[StepExecution]]:
        """Re-run only the baseline gates on the final tree: the probe is LLM-authored code and must not
        be able to change the sources after they were verified."""
        base_runner = VerificationRunner(
            steps=list(spec.verification_steps),
            allow_no_verify=spec.allow_no_verify,
            protected_paths=spec.protected_paths,
            allow_gate_edits=spec.allow_gate_edits,
            base_rev=spec.base_rev,
        )
        base_outcome = base_runner.run(sandbox, deadline=deadline)
        return base_outcome.passed, [replace(ex, step_id=f"{ex.step_id}:reverify") for ex in base_outcome.executions]

    try:
        outcome = runner.run(sandbox, deadline=deadline)
        executions = list(outcome.executions)
        attempts_used = 0
        telemetry: ModelTelemetry | None = None
        passed = outcome.passed

        if not passed:
            # Probe failed!
            remaining_retries = spec.max_healing_attempts - healing_attempts_so_far
            if remaining_retries <= 0:
                return False, 0, executions, None

            if healer is None:
                healer = SelfHealingLoop(max_retries=remaining_retries)

            repair_prompt = (
                f"ORIGINAL TASK:\n{spec.task_prompt}\n\n"
                "Your previous implementation passed baseline verification, but an adversarial red-team stress test "
                f"found defects:\n\n{outcome.error_summary}\n\n"
                "Fix the application code to satisfy the requirements and pass all tests (baseline and adversarial).\n"
                "Provide complete file contents formatted in ```file:<path> ... ``` blocks."
            )

            passed, attempts, healed_executions, telemetry, _streak = healer.run_loop(
                sandbox=sandbox,
                harness=harness,
                initial_prompt=repair_prompt,
                verification_runner=runner,
                target_files=spec.target_files or None,
                status_callback=status_callback,
                deadline=deadline,
            )
            executions += healed_executions
            attempts_used = attempts + 1

        if passed:
            cleanup_probe()
            baseline_ok, baseline_execs = reverify_baseline()
            executions += baseline_execs
            passed = baseline_ok
        return passed, attempts_used, executions, telemetry
    finally:
        cleanup_probe()

activity_apply_patch(repo_path, patch_content, target_branch=None, commit_msg=None)

Apply a verified patch to the repository and optionally commit it to a branch.

Atomic with respect to the host repo: refuses to touch files with uncommitted changes, and if applying or committing fails after a branch was created, the original branch is restored and the new one deleted.

Source code in dark_factory/orchestrator/activities.py

def activity_apply_patch(
    repo_path: str | Path,
    patch_content: str,
    target_branch: str | None = None,
    commit_msg: str | None = None,
) -> str:
    """Apply a verified patch to the repository and optionally commit it to a branch.

    Atomic with respect to the host repo: refuses to touch files with uncommitted changes, and if applying
    or committing fails after a branch was created, the original branch is restored and the new one deleted.
    """
    repo = Path(repo_path).resolve()
    if not patch_content.strip():
        raise DarkFactoryError("Cannot apply empty patch.")

    patch_files = parse_patch_files(patch_content)

    # 1. Never mix the operator's work-in-progress into the factory's commit
    if patch_files:
        dirty = _git(repo, "status", "--porcelain", "--", *patch_files, check=False).stdout.strip()
        if dirty:
            raise DarkFactoryError(f"Refusing to apply: target files have uncommitted changes:\n{dirty}")

    # 2. Pre-check patch applies cleanly before touching working tree or checking out branch
    check_proc = _git(
        repo,
        "apply",
        "--check",
        "--whitespace=nowarn",
        "--exclude=*.pyc",
        "--exclude=__pycache__/*",
        "-",
        check=False,
        input_text=patch_content,
    )
    if check_proc.returncode != 0:
        raise DarkFactoryError(f"Patch does not apply cleanly to target repository: {check_proc.stderr}")

    original_ref = _git(repo, "rev-parse", "--abbrev-ref", "HEAD", check=False).stdout.strip()
    if original_ref in ("", "HEAD"):  # detached HEAD
        original_ref = _git(repo, "rev-parse", "HEAD", check=False).stdout.strip()

    # If target_branch requested, checkout new branch
    if target_branch:
        try:
            _git(repo, "checkout", "-b", target_branch)
        except subprocess.CalledProcessError as e:
            raise _git_error(f"Creating branch '{target_branch}'", e) from e

    try:
        # 3. Apply patch via git apply
        apply_proc = _git(
            repo,
            "apply",
            "--whitespace=nowarn",
            "--exclude=*.pyc",
            "--exclude=__pycache__/*",
            "-",
            check=False,
            input_text=patch_content,
        )
        if apply_proc.returncode != 0:
            raise DarkFactoryError(f"Failed to apply patch: {apply_proc.stderr}")

        # 4. Stage ONLY modified files from patch (never indiscriminate git add .)
        if commit_msg:
            if patch_files:
                _git(repo, "add", "--", *patch_files)
            else:
                _git(repo, "add", "-u")

            _git(repo, "commit", "--no-gpg-sign", "-m", commit_msg)
            return _git(repo, "rev-parse", "HEAD").stdout.strip()
    except subprocess.CalledProcessError as e:
        _rollback_apply(repo, patch_files, original_ref, target_branch)
        raise _git_error("Committing patch", e) from e
    except DarkFactoryError:
        _rollback_apply(repo, patch_files, original_ref, target_branch)
        raise

    return "applied"

activity_cleanup_sandbox(sandbox)

Safely destroy sandbox.

Source code in dark_factory/orchestrator/activities.py

def activity_cleanup_sandbox(sandbox: Sandbox) -> None:
    """Safely destroy sandbox."""
    try:
        sandbox.destroy()
    except Exception:
        pass

activity_create_sandbox(spec, run_id, base_dir=None)

Create and isolate a git worktree sandbox at the specified revision.

Source code in dark_factory/orchestrator/activities.py

def activity_create_sandbox(
    spec: TaskSpec,
    run_id: str,
    base_dir: str | Path | None = None,
) -> GitWorktreeSandbox:
    """Create and isolate a git worktree sandbox at the specified revision."""
    sandbox = GitWorktreeSandbox(
        repo_path=spec.repo_path,
        sandbox_id=run_id,
        base_dir=base_dir,
    )
    sandbox.create(base_rev=spec.base_rev)
    return sandbox

activity_execute_task_and_verify(sandbox, harness, spec, status_callback=None, deadline=None, plan=None)

Execute code generation and deterministic verification with self-healing.

Source code in dark_factory/orchestrator/activities.py

def activity_execute_task_and_verify(
    sandbox: Sandbox,
    harness: AgentHarness,
    spec: TaskSpec,
    status_callback: Callable[[RunStatus], None] | None = None,
    deadline: float | None = None,
    plan: ExecutionPlan | None = None,
) -> tuple[bool, int, list[StepExecution], ModelTelemetry | None, int]:
    """Execute code generation and deterministic verification with self-healing."""
    runner = VerificationRunner(
        steps=spec.verification_steps,
        allow_no_verify=spec.allow_no_verify,
        protected_paths=spec.protected_paths,
        allow_gate_edits=spec.allow_gate_edits,
        base_rev=spec.base_rev,
    )
    healer = SelfHealingLoop(max_retries=spec.max_healing_attempts)

    initial_prompt = spec.task_prompt
    if plan and plan.summary:
        invariants_block = "\n".join(f"- {inv}" for inv in plan.invariants) if plan.invariants else "None specified"
        steps_block = "\n".join(f"- {step}" for step in plan.steps) if plan.steps else "None specified"
        initial_prompt = (
            f"{spec.task_prompt}\n\n"
            f"=== ARCHITECTURAL EXECUTION PLAN ===\n"
            f"Summary: {plan.summary}\n"
            f"Invariants:\n{invariants_block}\n"
            f"Implementation Steps:\n{steps_block}\n"
            f"Follow this plan strictly and make the minimal necessary changes."
        )

    passed, healing_attempts, executions, telemetry, repeated_failure_streak = healer.run_loop(
        sandbox=sandbox,
        harness=harness,
        initial_prompt=initial_prompt,
        verification_runner=runner,
        target_files=spec.target_files or None,
        status_callback=status_callback,
        deadline=deadline,
    )
    return passed, healing_attempts, executions, telemetry, repeated_failure_streak

activity_plan_task(run_id, spec, sandbox, manifest, harness=None, deadline=None)

Execute the reasoning planner to produce an ExecutionPlan before code generation.

Source code in dark_factory/orchestrator/activities.py

def activity_plan_task(
    run_id: str,
    spec: TaskSpec,
    sandbox: Sandbox,
    manifest: EvidenceManifest,
    harness: AgentHarness | None = None,
    deadline: float | None = None,
) -> ExecutionPlan | None:
    """Execute the reasoning planner to produce an ExecutionPlan before code generation."""
    if spec.skip_plan:
        return None
    if deadline is not None and time.monotonic() >= deadline:
        raise RunTimeoutError("Run deadline exceeded before task planning.")

    from dark_factory.harness.local_coder import LocalCoderHarness
    from dark_factory.planning.planner import LocalPlanner

    planner_model = spec.planner_model or spec.model
    if isinstance(harness, LocalCoderHarness) and (spec.planner_model is None or harness.model == planner_model):
        planner_harness = harness
    elif isinstance(harness, LocalCoderHarness):
        planner_harness = harness.with_model(planner_model)
    elif harness is not None and not hasattr(harness, "_call_model"):
        plan = ExecutionPlan(
            plan_id=f"plan-{run_id}",
            summary="Deterministic execution plan",
            invariants=[],
            steps=[],
            target_files=[],
        )
        manifest.execution_plan = plan
        return plan
    else:
        planner_harness = LocalCoderHarness(model=planner_model)

    planner = LocalPlanner(harness=planner_harness)

    # Lightweight repository context
    context = ""
    try:
        res = sandbox.execute(["git", "ls-files"])
        if res.passed and res.stdout.strip():
            context = f"Files in repository:\n{res.stdout.strip()[:1000]}"
    except Exception:
        pass

    with _clamped_harness(planner, deadline):
        plan = planner.generate_plan(
            task_prompt=spec.task_prompt,
            context=context,
            plan_id=f"plan-{run_id}",
        )
    manifest.execution_plan = plan
    return plan

activity_preserve_evidence(sandbox, locker, manifest, executions, healing_attempts, telemetry, diff=None)

Extract patch (unless already extracted), assemble audit transcript, and preserve in evidence locker.

Source code in dark_factory/orchestrator/activities.py

def activity_preserve_evidence(
    sandbox: Sandbox,
    locker: EvidenceLocker,
    manifest: EvidenceManifest,
    executions: list[StepExecution],
    healing_attempts: int,
    telemetry: ModelTelemetry | None,
    diff: str | None = None,
) -> str:
    """Extract patch (unless already extracted), assemble audit transcript, and preserve in evidence locker."""
    if diff is None:
        diff = sandbox.get_diff()

    # Build transcript
    transcript_lines = [f"=== RUN {manifest.run_id} TRANSCRIPT ==="]
    for ex in executions:
        status_str = "PASSED" if ex.passed else f"FAILED (exit {ex.exit_code})"
        transcript_lines.append(f"[{ex.step_id}] {status_str} ({ex.duration_sec:.2f}s)")
        if ex.stdout.strip():
            transcript_lines.append(f"STDOUT:\n{ex.stdout.strip()}")
        if ex.stderr.strip():
            transcript_lines.append(f"STDERR:\n{ex.stderr.strip()}")

    manifest.verification_results = executions
    manifest.healing_attempts = healing_attempts
    manifest.model_telemetry = telemetry
    manifest.completed_at = datetime.now(UTC).isoformat()

    locker.save_run(
        manifest=manifest,
        patch_content=diff,
        transcript="\n".join(transcript_lines),
    )
    return diff

parse_patch_files(patch_content)

Extract list of target file paths from git unified diff (spaces and quoted paths supported).

Source code in dark_factory/orchestrator/activities.py

def parse_patch_files(patch_content: str) -> list[str]:
    """Extract list of target file paths from git unified diff (spaces and quoted paths supported)."""
    files = set()
    for line in patch_content.splitlines():
        if line.startswith("diff --git "):
            path = _header_path(line.removeprefix("diff --git "))
            if path:
                files.add(path)
    return sorted(files)