Skip to content

Verification & Self-Healing

Deterministic gate execution, tamper prevention, and feedback loop.

dark_factory.verification.runner.VerificationOutcome dataclass

Outcome of running a suite of verification gates.

Source code in dark_factory/verification/runner.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@dataclass
class VerificationOutcome:
    """Outcome of running a suite of verification gates."""

    passed: bool
    executions: list[StepExecution] = field(default_factory=list)
    failed_step: StepExecution | None = None
    tampered_paths: list[str] = field(default_factory=list)

    @property
    def error_summary(self) -> str:
        if not self.failed_step:
            return ""
        stderr = self.failed_step.stderr.strip()
        stdout = self.failed_step.stdout.strip()
        msg = f"Gate '{self.failed_step.step_id}' failed with exit code {self.failed_step.exit_code}."
        if stderr:
            msg += f"\nSTDERR:\n{stderr}"
        elif stdout:
            msg += f"\nSTDOUT:\n{stdout}"
        return msg

dark_factory.verification.runner.VerificationRunner

Executes structured verification steps sequentially inside a sandbox.

Source code in dark_factory/verification/runner.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class VerificationRunner:
    """Executes structured verification steps sequentially inside a sandbox."""

    def __init__(
        self,
        steps: list[VerificationStep],
        allow_no_verify: bool = False,
        protected_paths: list[str] | None = None,
        allow_gate_edits: bool = False,
        base_rev: str = "HEAD",
    ) -> None:
        self.steps = steps
        self.allow_no_verify = allow_no_verify
        self.protected_paths = protected_paths or []
        self.allow_gate_edits = allow_gate_edits
        self.base_rev = base_rev

    def detect_default_gate_paths(self) -> list[str]:
        """Infer default protected gate files and directories from configured steps.

        A step that invokes a linter/formatter/type-checker (`ruff`, `mypy`, ...) names its lint
        *targets* in argv, not test/gate files — those are exactly the files the agent is expected to
        edit, so such a step contributes nothing to the detected set (the fixed `_DEFAULT_PROTECTED_PATHS`
        baseline still applies). Without this, a gate like `ruff check <file>` would get `<file>` silently
        reverted to baseline before every verification pass, making the task unsolvable whenever the lint
        target is the very file the agent must fix.
        """
        paths = set()
        for step in self.steps:
            if any(Path(arg.removeprefix("./")).stem in _LINTER_LIKE_TOOLS for arg in step.argv):
                continue
            for arg in step.argv:
                arg_clean = arg.removeprefix("./")
                if (
                    arg_clean
                    and not arg_clean.startswith("-")
                    and arg_clean not in _RUNNER_LITERALS
                    and not arg_clean.endswith(".exe")
                    and arg_clean != "test_adversarial_probe.py"
                ):
                    paths.add(arg_clean)
        paths.update(_DEFAULT_PROTECTED_PATHS)
        paths.update(f":(glob)**/{name}" for name in _DEFAULT_PROTECTED_ANYWHERE)
        return sorted(paths)

    def detect_runner_paths(self) -> list[str]:
        modules = set()
        for step in self.steps:
            argv = step.argv
            if not argv:
                continue
            executable = Path(argv[0]).name
            if executable.startswith("python") and "-m" in argv:
                i = argv.index("-m") + 1
                if i < len(argv) and all(part.isidentifier() for part in argv[i].split(".")):
                    modules.add(argv[i].split(".")[0])
            elif Path(argv[0]).stem in {"pytest", "ruff", "flake8", "pylint", "mypy", "pyright", "black", "isort"}:
                modules.add(Path(argv[0]).stem)
        if "pytest" in modules:
            modules.update({"_pytest", "pluggy"})
        paths = set()
        for name in modules:
            paths.update({name, name + ".py", name + ".pyc", f":(glob)__pycache__/{name}.*.pyc"})
        return sorted(paths)

    def run(self, sandbox: Sandbox, deadline: float | None = None) -> VerificationOutcome:
        """Run all verification steps. Stops on the first failing mandatory step.

        `deadline` is a `time.monotonic()` instant; each gate's timeout is clamped to the time remaining.
        """
        if not self.steps:
            if self.allow_no_verify:
                return VerificationOutcome(passed=True, executions=[])
            return VerificationOutcome(
                passed=False,
                executions=[],
                failed_step=StepExecution(
                    step_id="verification_gate_check",
                    exit_code=1,
                    stdout="",
                    stderr=(
                        "Zero verification gates configured. A dark factory run requires deterministic "
                        "verification gates (or explicit --no-verify)."
                    ),
                    duration_sec=0.0,
                ),
            )

        # Gate protection: restore baseline test/gate files before running verification
        restored: list[str] = []
        if not self.allow_gate_edits and hasattr(sandbox, "restore_paths"):
            paths_to_protect = self.protected_paths or self.detect_default_gate_paths()
            restored = sandbox.restore_paths(paths_to_protect, rev=self.base_rev)
            restored = sorted(
                set(restored + sandbox.restore_runner_paths(self.detect_runner_paths(), rev=self.base_rev))
            )

        executions = []

        for step in self.steps:
            # Spec §202: deadline is enforced at every phase boundary. When it has already
            # passed, the next gate must NOT be launched; we surface the partial executions
            # so the engine can adopt a TIMED_OUT outcome without spawning further sub-processes.
            if deadline is not None and time.monotonic() >= deadline:
                raise RunTimeoutError(
                    "Verification deadline exceeded before gate execution.",
                    executions=executions,
                )
            timeout = step.timeout_sec
            if deadline is not None:
                remaining = int(deadline - time.monotonic())
                if remaining <= 0:
                    # Defensive: should be unreachable thanks to the short-circuit above.
                    raise RunTimeoutError(
                        "Verification deadline exceeded during gate scheduling.",
                        executions=executions,
                    )
                timeout = min(timeout, remaining)
            exec_res = sandbox.execute(
                argv=step.argv,
                timeout=timeout,
                step_id=step.id,
            )
            executions.append(exec_res)

            if step.mandatory and not exec_res.passed:
                return VerificationOutcome(
                    passed=False,
                    executions=executions,
                    failed_step=exec_res,
                    tampered_paths=restored,
                )

        return VerificationOutcome(
            passed=True,
            executions=executions,
            tampered_paths=restored,
        )

Methods:

detect_default_gate_paths()

Infer default protected gate files and directories from configured steps.

A step that invokes a linter/formatter/type-checker (ruff, mypy, ...) names its lint targets in argv, not test/gate files — those are exactly the files the agent is expected to edit, so such a step contributes nothing to the detected set (the fixed _DEFAULT_PROTECTED_PATHS baseline still applies). Without this, a gate like ruff check <file> would get <file> silently reverted to baseline before every verification pass, making the task unsolvable whenever the lint target is the very file the agent must fix.

Source code in dark_factory/verification/runner.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def detect_default_gate_paths(self) -> list[str]:
    """Infer default protected gate files and directories from configured steps.

    A step that invokes a linter/formatter/type-checker (`ruff`, `mypy`, ...) names its lint
    *targets* in argv, not test/gate files — those are exactly the files the agent is expected to
    edit, so such a step contributes nothing to the detected set (the fixed `_DEFAULT_PROTECTED_PATHS`
    baseline still applies). Without this, a gate like `ruff check <file>` would get `<file>` silently
    reverted to baseline before every verification pass, making the task unsolvable whenever the lint
    target is the very file the agent must fix.
    """
    paths = set()
    for step in self.steps:
        if any(Path(arg.removeprefix("./")).stem in _LINTER_LIKE_TOOLS for arg in step.argv):
            continue
        for arg in step.argv:
            arg_clean = arg.removeprefix("./")
            if (
                arg_clean
                and not arg_clean.startswith("-")
                and arg_clean not in _RUNNER_LITERALS
                and not arg_clean.endswith(".exe")
                and arg_clean != "test_adversarial_probe.py"
            ):
                paths.add(arg_clean)
    paths.update(_DEFAULT_PROTECTED_PATHS)
    paths.update(f":(glob)**/{name}" for name in _DEFAULT_PROTECTED_ANYWHERE)
    return sorted(paths)

run(sandbox, deadline=None)

Run all verification steps. Stops on the first failing mandatory step.

deadline is a time.monotonic() instant; each gate's timeout is clamped to the time remaining.

Source code in dark_factory/verification/runner.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def run(self, sandbox: Sandbox, deadline: float | None = None) -> VerificationOutcome:
    """Run all verification steps. Stops on the first failing mandatory step.

    `deadline` is a `time.monotonic()` instant; each gate's timeout is clamped to the time remaining.
    """
    if not self.steps:
        if self.allow_no_verify:
            return VerificationOutcome(passed=True, executions=[])
        return VerificationOutcome(
            passed=False,
            executions=[],
            failed_step=StepExecution(
                step_id="verification_gate_check",
                exit_code=1,
                stdout="",
                stderr=(
                    "Zero verification gates configured. A dark factory run requires deterministic "
                    "verification gates (or explicit --no-verify)."
                ),
                duration_sec=0.0,
            ),
        )

    # Gate protection: restore baseline test/gate files before running verification
    restored: list[str] = []
    if not self.allow_gate_edits and hasattr(sandbox, "restore_paths"):
        paths_to_protect = self.protected_paths or self.detect_default_gate_paths()
        restored = sandbox.restore_paths(paths_to_protect, rev=self.base_rev)
        restored = sorted(
            set(restored + sandbox.restore_runner_paths(self.detect_runner_paths(), rev=self.base_rev))
        )

    executions = []

    for step in self.steps:
        # Spec §202: deadline is enforced at every phase boundary. When it has already
        # passed, the next gate must NOT be launched; we surface the partial executions
        # so the engine can adopt a TIMED_OUT outcome without spawning further sub-processes.
        if deadline is not None and time.monotonic() >= deadline:
            raise RunTimeoutError(
                "Verification deadline exceeded before gate execution.",
                executions=executions,
            )
        timeout = step.timeout_sec
        if deadline is not None:
            remaining = int(deadline - time.monotonic())
            if remaining <= 0:
                # Defensive: should be unreachable thanks to the short-circuit above.
                raise RunTimeoutError(
                    "Verification deadline exceeded during gate scheduling.",
                    executions=executions,
                )
            timeout = min(timeout, remaining)
        exec_res = sandbox.execute(
            argv=step.argv,
            timeout=timeout,
            step_id=step.id,
        )
        executions.append(exec_res)

        if step.mandatory and not exec_res.passed:
            return VerificationOutcome(
                passed=False,
                executions=executions,
                failed_step=exec_res,
                tampered_paths=restored,
            )

    return VerificationOutcome(
        passed=True,
        executions=executions,
        tampered_paths=restored,
    )

dark_factory.verification.healing.SelfHealingLoop

Orchestrates the agent-verification feedback loop up to max_retries.

Source code in dark_factory/verification/healing.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class SelfHealingLoop:
    """Orchestrates the agent-verification feedback loop up to max_retries."""

    def __init__(self, max_retries: int = 3) -> None:
        self.max_retries = max_retries

    def run_loop(
        self,
        sandbox: Sandbox,
        harness: AgentHarness,
        initial_prompt: str,
        verification_runner: VerificationRunner,
        target_files: list[str] | None = None,
        status_callback: Callable[[RunStatus], None] | None = None,
        deadline: float | None = None,
    ) -> tuple[bool, int, list[StepExecution], ModelTelemetry | None, int]:
        """Run initial generation and up to max_retries self-healing attempts.

        `deadline` is a `time.monotonic()` instant checked at every phase boundary; when it has passed a
        `RunTimeoutError` carrying the partial results is raised.

        Returns:
            (passed, total_healing_attempts, all_executions, aggregated_telemetry, repeated_failure_streak)

        `repeated_failure_streak` counts consecutive verification failures with an identical outcome
        (same gate, exit code, stdout, stderr) ending at the return point — a model that keeps re-deriving
        the same wrong fix produces byte-identical verification failures attempt after attempt, which is a
        distinct failure mode from trying different (still wrong) approaches each time. It's 0 on success.
        """

        def check_deadline() -> None:
            if deadline is not None and time.monotonic() >= deadline:
                raise RunTimeoutError(
                    "Run deadline exceeded.",
                    executions=all_executions,
                    healing_attempts=healing_attempts,
                    telemetry=aggregated_telemetry,
                )

        all_executions: list[StepExecution] = []
        healing_attempts = 0
        current_prompt = initial_prompt
        aggregated_telemetry: ModelTelemetry | None = None
        last_failure_signature: tuple[str, int, str, str] | None = None
        repeated_failure_streak = 0

        while True:
            check_deadline()

            # 1. Run Harness
            harness_result = harness.execute_task(
                sandbox=sandbox,
                task_prompt=current_prompt,
                target_files=target_files,
            )

            # Aggregate telemetry across attempts
            if harness_result.telemetry:
                t = harness_result.telemetry
                if aggregated_telemetry is None:
                    aggregated_telemetry = ModelTelemetry(
                        engine=t.engine,
                        model_name=t.model_name,
                        prompt_tokens=t.prompt_tokens,
                        completion_tokens=t.completion_tokens,
                        total_tokens=t.total_tokens,
                        duration_sec=t.duration_sec,
                        tokens_per_sec=t.tokens_per_sec,
                        cost_usd=0.0,
                        num_ctx=t.num_ctx,
                        finish_reason=t.finish_reason,
                        max_tokens=t.max_tokens,
                        reasoning_chars=t.reasoning_chars,
                    )
                else:
                    aggregated_telemetry.prompt_tokens += t.prompt_tokens
                    aggregated_telemetry.completion_tokens += t.completion_tokens
                    aggregated_telemetry.total_tokens += t.total_tokens
                    aggregated_telemetry.duration_sec = round(aggregated_telemetry.duration_sec + t.duration_sec, 3)
                    if t.num_ctx is not None:  # report the largest window any attempt used
                        aggregated_telemetry.num_ctx = max(aggregated_telemetry.num_ctx or 0, t.num_ctx)
                    if t.max_tokens is not None:  # likewise the largest completion budget
                        aggregated_telemetry.max_tokens = max(aggregated_telemetry.max_tokens or 0, t.max_tokens)
                    aggregated_telemetry.finish_reason = t.finish_reason  # of the last call
                    aggregated_telemetry.reasoning_chars += t.reasoning_chars
                    if aggregated_telemetry.duration_sec > 0:
                        aggregated_telemetry.tokens_per_sec = round(
                            aggregated_telemetry.completion_tokens / aggregated_telemetry.duration_sec, 2
                        )

            elif aggregated_telemetry is not None:
                aggregated_telemetry.finish_reason = None  # the last call reported nothing

            # If harness failed to generate or parse blocks, treat as a healing retry
            if not harness_result.success:
                error_msg = harness_result.error or "Model did not output any recognizable file blocks."
                if healing_attempts >= self.max_retries:
                    # No gate ran, so keep the reason in the evidence instead of an unexplained FAILED. The
                    # last failure was the harness's, so an earlier gate-failure streak no longer applies.
                    all_executions.append(StepExecution("harness", 1, "", error_msg, 0.0))
                    return False, healing_attempts, all_executions, aggregated_telemetry, 0

                healing_attempts += 1

                if status_callback:
                    status_callback(RunStatus.SELF_HEALING)

                if harness_result.telemetry and harness_result.telemetry.finish_reason == "length":
                    # The same request would hit the same token limit. Nothing was applied, so every needed
                    # file must be re-sent: ask for a compact reply. (The error text is for the operator only.)
                    current_prompt = (
                        f"ORIGINAL TASK:\n{initial_prompt}\n\n"
                        f"REPAIR ATTEMPT #{healing_attempts} OF {self.max_retries}:\n"
                        "Your previous reply was cut off by the token limit and no file was written.\n\n"
                        "Reply again with each file that must change as a complete ```file:<path> ... ``` block. "
                        "Keep it compact: no explanation, no reasoning text, no comments beyond what the code needs."
                    )
                    continue

                current_prompt = (
                    f"ORIGINAL TASK:\n{initial_prompt}\n\n"
                    f"REPAIR ATTEMPT #{healing_attempts} OF {self.max_retries}:\n"
                    f"Your previous response failed to apply: {error_msg}\n\n"
                    "Please provide complete file contents formatted inside code blocks: ```file:<path> ... ```."
                )
                continue

            # 2. Run Verification
            check_deadline()
            if status_callback:
                status_callback(RunStatus.VERIFYING)

            outcome: VerificationOutcome = verification_runner.run(sandbox, deadline=deadline)
            all_executions.extend(outcome.executions)

            # If all verification gates passed, we succeed!
            if outcome.passed:
                return True, healing_attempts, all_executions, aggregated_telemetry, 0

            # Detect a model stuck re-deriving the same wrong fix: identical gate, exit code, and
            # stdout/stderr as the immediately preceding attempt (duration/timed_out excluded, and any
            # embedded "N.NNs" duration substring normalized away — e.g. pytest's own "1 failed in 0.03s"
            # footer — since those vary run to run even when the underlying code and outcome are unchanged).
            failed = outcome.failed_step
            failure_signature = (
                (
                    failed.step_id,
                    failed.exit_code,
                    _normalize_for_signature(failed.stdout),
                    _normalize_for_signature(failed.stderr),
                )
                if failed
                else None
            )
            if failure_signature is not None and failure_signature == last_failure_signature:
                repeated_failure_streak += 1
            else:
                repeated_failure_streak = 0
            last_failure_signature = failure_signature

            # 3. Check retry budget
            if healing_attempts >= self.max_retries:
                # Exceeded retry budget
                return False, healing_attempts, all_executions, aggregated_telemetry, repeated_failure_streak

            healing_attempts += 1
            if status_callback:
                status_callback(RunStatus.SELF_HEALING)

            # 4. Construct repair prompt with error trace AND original task retained
            # A full pytest dump can exceed the model's window; the head names the gate, the tail has the summary.
            error_trace, _ = cap_text(
                outcome.error_summary, MAX_ERROR_TRACE_CHARS, keep_tail=MAX_ERROR_TRACE_CHARS * 3 // 4
            )
            tamper_notice = ""
            if outcome.tampered_paths:
                tamper_notice = (
                    f"SECURITY NOTICE: Modifications to protected verification files "
                    f"({', '.join(outcome.tampered_paths)}) were detected and REVERTED to baseline.\n"
                    "You must NOT modify verification scripts or test files. Fix the application code instead!\n\n"
                )
            stuck_notice = ""
            if repeated_failure_streak >= 1:
                stuck_notice = (
                    "STUCK NOTICE: Your last fix resulted in the exact same test failure as the attempt "
                    "before it — your current approach is not working. Do not make a small tweak to the "
                    "same logic; use a fundamentally different algorithm or data structure to solve this "
                    "problem.\n\n"
                )

            current_prompt = (
                f"ORIGINAL TASK:\n{initial_prompt}\n\n"
                f"REPAIR ATTEMPT #{healing_attempts} OF {self.max_retries}:\n"
                f"Your previous changes failed automated deterministic verification.\n"
                f"{tamper_notice}"
                f"{stuck_notice}"
                f"{error_trace}\n\n"
                "Please analyze the errors in the context of the original task and generate the corrected full file contents.\n"
                "Remember to format all updated files in ```file:<path> ... ``` blocks."
            )

Methods:

run_loop(sandbox, harness, initial_prompt, verification_runner, target_files=None, status_callback=None, deadline=None)

Run initial generation and up to max_retries self-healing attempts.

deadline is a time.monotonic() instant checked at every phase boundary; when it has passed a RunTimeoutError carrying the partial results is raised.

Returns:

Type Description
tuple[bool, int, list[StepExecution], ModelTelemetry | None, int]

(passed, total_healing_attempts, all_executions, aggregated_telemetry, repeated_failure_streak)

repeated_failure_streak counts consecutive verification failures with an identical outcome (same gate, exit code, stdout, stderr) ending at the return point — a model that keeps re-deriving the same wrong fix produces byte-identical verification failures attempt after attempt, which is a distinct failure mode from trying different (still wrong) approaches each time. It's 0 on success.

Source code in dark_factory/verification/healing.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def run_loop(
    self,
    sandbox: Sandbox,
    harness: AgentHarness,
    initial_prompt: str,
    verification_runner: VerificationRunner,
    target_files: list[str] | None = None,
    status_callback: Callable[[RunStatus], None] | None = None,
    deadline: float | None = None,
) -> tuple[bool, int, list[StepExecution], ModelTelemetry | None, int]:
    """Run initial generation and up to max_retries self-healing attempts.

    `deadline` is a `time.monotonic()` instant checked at every phase boundary; when it has passed a
    `RunTimeoutError` carrying the partial results is raised.

    Returns:
        (passed, total_healing_attempts, all_executions, aggregated_telemetry, repeated_failure_streak)

    `repeated_failure_streak` counts consecutive verification failures with an identical outcome
    (same gate, exit code, stdout, stderr) ending at the return point — a model that keeps re-deriving
    the same wrong fix produces byte-identical verification failures attempt after attempt, which is a
    distinct failure mode from trying different (still wrong) approaches each time. It's 0 on success.
    """

    def check_deadline() -> None:
        if deadline is not None and time.monotonic() >= deadline:
            raise RunTimeoutError(
                "Run deadline exceeded.",
                executions=all_executions,
                healing_attempts=healing_attempts,
                telemetry=aggregated_telemetry,
            )

    all_executions: list[StepExecution] = []
    healing_attempts = 0
    current_prompt = initial_prompt
    aggregated_telemetry: ModelTelemetry | None = None
    last_failure_signature: tuple[str, int, str, str] | None = None
    repeated_failure_streak = 0

    while True:
        check_deadline()

        # 1. Run Harness
        harness_result = harness.execute_task(
            sandbox=sandbox,
            task_prompt=current_prompt,
            target_files=target_files,
        )

        # Aggregate telemetry across attempts
        if harness_result.telemetry:
            t = harness_result.telemetry
            if aggregated_telemetry is None:
                aggregated_telemetry = ModelTelemetry(
                    engine=t.engine,
                    model_name=t.model_name,
                    prompt_tokens=t.prompt_tokens,
                    completion_tokens=t.completion_tokens,
                    total_tokens=t.total_tokens,
                    duration_sec=t.duration_sec,
                    tokens_per_sec=t.tokens_per_sec,
                    cost_usd=0.0,
                    num_ctx=t.num_ctx,
                    finish_reason=t.finish_reason,
                    max_tokens=t.max_tokens,
                    reasoning_chars=t.reasoning_chars,
                )
            else:
                aggregated_telemetry.prompt_tokens += t.prompt_tokens
                aggregated_telemetry.completion_tokens += t.completion_tokens
                aggregated_telemetry.total_tokens += t.total_tokens
                aggregated_telemetry.duration_sec = round(aggregated_telemetry.duration_sec + t.duration_sec, 3)
                if t.num_ctx is not None:  # report the largest window any attempt used
                    aggregated_telemetry.num_ctx = max(aggregated_telemetry.num_ctx or 0, t.num_ctx)
                if t.max_tokens is not None:  # likewise the largest completion budget
                    aggregated_telemetry.max_tokens = max(aggregated_telemetry.max_tokens or 0, t.max_tokens)
                aggregated_telemetry.finish_reason = t.finish_reason  # of the last call
                aggregated_telemetry.reasoning_chars += t.reasoning_chars
                if aggregated_telemetry.duration_sec > 0:
                    aggregated_telemetry.tokens_per_sec = round(
                        aggregated_telemetry.completion_tokens / aggregated_telemetry.duration_sec, 2
                    )

        elif aggregated_telemetry is not None:
            aggregated_telemetry.finish_reason = None  # the last call reported nothing

        # If harness failed to generate or parse blocks, treat as a healing retry
        if not harness_result.success:
            error_msg = harness_result.error or "Model did not output any recognizable file blocks."
            if healing_attempts >= self.max_retries:
                # No gate ran, so keep the reason in the evidence instead of an unexplained FAILED. The
                # last failure was the harness's, so an earlier gate-failure streak no longer applies.
                all_executions.append(StepExecution("harness", 1, "", error_msg, 0.0))
                return False, healing_attempts, all_executions, aggregated_telemetry, 0

            healing_attempts += 1

            if status_callback:
                status_callback(RunStatus.SELF_HEALING)

            if harness_result.telemetry and harness_result.telemetry.finish_reason == "length":
                # The same request would hit the same token limit. Nothing was applied, so every needed
                # file must be re-sent: ask for a compact reply. (The error text is for the operator only.)
                current_prompt = (
                    f"ORIGINAL TASK:\n{initial_prompt}\n\n"
                    f"REPAIR ATTEMPT #{healing_attempts} OF {self.max_retries}:\n"
                    "Your previous reply was cut off by the token limit and no file was written.\n\n"
                    "Reply again with each file that must change as a complete ```file:<path> ... ``` block. "
                    "Keep it compact: no explanation, no reasoning text, no comments beyond what the code needs."
                )
                continue

            current_prompt = (
                f"ORIGINAL TASK:\n{initial_prompt}\n\n"
                f"REPAIR ATTEMPT #{healing_attempts} OF {self.max_retries}:\n"
                f"Your previous response failed to apply: {error_msg}\n\n"
                "Please provide complete file contents formatted inside code blocks: ```file:<path> ... ```."
            )
            continue

        # 2. Run Verification
        check_deadline()
        if status_callback:
            status_callback(RunStatus.VERIFYING)

        outcome: VerificationOutcome = verification_runner.run(sandbox, deadline=deadline)
        all_executions.extend(outcome.executions)

        # If all verification gates passed, we succeed!
        if outcome.passed:
            return True, healing_attempts, all_executions, aggregated_telemetry, 0

        # Detect a model stuck re-deriving the same wrong fix: identical gate, exit code, and
        # stdout/stderr as the immediately preceding attempt (duration/timed_out excluded, and any
        # embedded "N.NNs" duration substring normalized away — e.g. pytest's own "1 failed in 0.03s"
        # footer — since those vary run to run even when the underlying code and outcome are unchanged).
        failed = outcome.failed_step
        failure_signature = (
            (
                failed.step_id,
                failed.exit_code,
                _normalize_for_signature(failed.stdout),
                _normalize_for_signature(failed.stderr),
            )
            if failed
            else None
        )
        if failure_signature is not None and failure_signature == last_failure_signature:
            repeated_failure_streak += 1
        else:
            repeated_failure_streak = 0
        last_failure_signature = failure_signature

        # 3. Check retry budget
        if healing_attempts >= self.max_retries:
            # Exceeded retry budget
            return False, healing_attempts, all_executions, aggregated_telemetry, repeated_failure_streak

        healing_attempts += 1
        if status_callback:
            status_callback(RunStatus.SELF_HEALING)

        # 4. Construct repair prompt with error trace AND original task retained
        # A full pytest dump can exceed the model's window; the head names the gate, the tail has the summary.
        error_trace, _ = cap_text(
            outcome.error_summary, MAX_ERROR_TRACE_CHARS, keep_tail=MAX_ERROR_TRACE_CHARS * 3 // 4
        )
        tamper_notice = ""
        if outcome.tampered_paths:
            tamper_notice = (
                f"SECURITY NOTICE: Modifications to protected verification files "
                f"({', '.join(outcome.tampered_paths)}) were detected and REVERTED to baseline.\n"
                "You must NOT modify verification scripts or test files. Fix the application code instead!\n\n"
            )
        stuck_notice = ""
        if repeated_failure_streak >= 1:
            stuck_notice = (
                "STUCK NOTICE: Your last fix resulted in the exact same test failure as the attempt "
                "before it — your current approach is not working. Do not make a small tweak to the "
                "same logic; use a fundamentally different algorithm or data structure to solve this "
                "problem.\n\n"
            )

        current_prompt = (
            f"ORIGINAL TASK:\n{initial_prompt}\n\n"
            f"REPAIR ATTEMPT #{healing_attempts} OF {self.max_retries}:\n"
            f"Your previous changes failed automated deterministic verification.\n"
            f"{tamper_notice}"
            f"{stuck_notice}"
            f"{error_trace}\n\n"
            "Please analyze the errors in the context of the original task and generate the corrected full file contents.\n"
            "Remember to format all updated files in ```file:<path> ... ``` blocks."
        )