Skip to content

naqsha.reflection

Reflection Loop and Automated Rollback Manager

naqsha.reflection

Reflection Loop boundaries.

RELIABILITY_GATE_TEST_PATHS module-attribute 

RELIABILITY_GATE_TEST_PATHS: tuple[str, ...] = (
    "tests/test_trace_replay.py",
    "tests/test_protocols.py",
    "tests/test_policy_and_trace.py",
    "tests/test_memory_simplemem_cross.py",
    "tests/redteam/test_corpus.py",
)

ReflectionLoop

Bases: Protocol

Source code in src/naqsha/reflection/base.py 
61
62
63
class ReflectionLoop(Protocol):
    def propose_patch(self, trace: list[TraceEvent]) -> ReflectionPatch | None:
        """Create an isolated patch candidate, never a runtime hotpatch."""

propose_patch 

propose_patch(
    trace: list[TraceEvent],
) -> ReflectionPatch | None

Create an isolated patch candidate, never a runtime hotpatch.

Source code in src/naqsha/reflection/base.py 
62
63
def propose_patch(self, trace: list[TraceEvent]) -> ReflectionPatch | None:
    """Create an isolated patch candidate, never a runtime hotpatch."""

ReflectionPatch dataclass

Artifacts for human review.

Merging into the active workspace is performed only by AutomatedRollbackManager inside the reflection package when workspace policy allows auto-merge; there is no merge method on this object.

Source code in src/naqsha/reflection/base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@dataclass(frozen=True)
class ReflectionPatch:
    """Artifacts for human review.

    Merging into the active workspace is performed only by ``AutomatedRollbackManager``
    inside the reflection package when workspace policy allows auto-merge; there is no
    merge method on this object.
    """

    workspace: Path
    summary: str
    reliability_gate_passed: bool
    auto_merged: bool = False

    @property
    def ready_for_human_review(self) -> bool:
        """True when the Reliability Gate passed; merge still needs human approval."""

        return self.reliability_gate_passed

workspace instance-attribute 

workspace: Path

summary instance-attribute 

summary: str

reliability_gate_passed instance-attribute 

reliability_gate_passed: bool

auto_merged class-attribute instance-attribute 

auto_merged: bool = False

ready_for_human_review property 

ready_for_human_review: bool

True when the Reliability Gate passed; merge still needs human approval.

ReflectionPatchEventSink

Bases: Protocol

Callbacks for patch lifecycle (implemented by embedders using RuntimeEventBus).

Source code in src/naqsha/reflection/base.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class ReflectionPatchEventSink(Protocol):
    """Callbacks for patch lifecycle (implemented by embedders using ``RuntimeEventBus``)."""

    def patch_merged(
        self,
        *,
        run_id: str,
        agent_id: str,
        patch_id: str,
        auto_merged: bool,
    ) -> None:
        """Patch merged into the team workspace after Reliability Gate pass."""

    def patch_rolled_back(
        self,
        *,
        run_id: str,
        agent_id: str,
        patch_id: str,
        reason: str,
    ) -> None:
        """Workspace restored after a failed post-merge boot check."""

patch_merged 

patch_merged(
    *,
    run_id: str,
    agent_id: str,
    patch_id: str,
    auto_merged: bool,
) -> None

Patch merged into the team workspace after Reliability Gate pass.

Source code in src/naqsha/reflection/base.py 
40
41
42
43
44
45
46
47
48
def patch_merged(
    self,
    *,
    run_id: str,
    agent_id: str,
    patch_id: str,
    auto_merged: bool,
) -> None:
    """Patch merged into the team workspace after Reliability Gate pass."""

patch_rolled_back 

patch_rolled_back(
    *,
    run_id: str,
    agent_id: str,
    patch_id: str,
    reason: str,
) -> None

Workspace restored after a failed post-merge boot check.

Source code in src/naqsha/reflection/base.py 
50
51
52
53
54
55
56
57
58
def patch_rolled_back(
    self,
    *,
    run_id: str,
    agent_id: str,
    patch_id: str,
    reason: str,
) -> None:
    """Workspace restored after a failed post-merge boot check."""

ReflectionTomlSettings dataclass

Subset of [reflection] used by SimpleReflectionLoop.

reliability_gate does not skip pytest for patch eligibility: it only gates whether auto_merge may run after the gate passes (fail-closed for autonomous merges).

Source code in src/naqsha/reflection/config.py 
11
12
13
14
15
16
17
18
19
20
21
@dataclass(frozen=True)
class ReflectionTomlSettings:
    """Subset of ``[reflection]`` used by ``SimpleReflectionLoop``.

    ``reliability_gate`` does **not** skip pytest for patch eligibility: it only gates whether
    ``auto_merge`` may run after the gate passes (fail-closed for autonomous merges).
    """

    enabled: bool = False
    auto_merge: bool = False
    reliability_gate: bool = True

enabled class-attribute instance-attribute 

enabled: bool = False

auto_merge class-attribute instance-attribute 

auto_merge: bool = False

reliability_gate class-attribute instance-attribute 

reliability_gate: bool = True

SimpleReflectionLoop

Builds isolated Reflection Patch folders from a QAOA trace.

This class does not import or call Tool Policy, the Core Runtime, or model adapters.

Source code in src/naqsha/reflection/loop.py 
 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
class SimpleReflectionLoop:
    """Builds isolated Reflection Patch folders from a QAOA trace.

    This class does not import or call Tool Policy, the Core Runtime, or model adapters.
    """

    def __init__(
        self,
        *,
        workspace_parent: Path,
        team_workspace: Path | None = None,
        project_root: Path | None = None,
        gate_runner: GateRunner | None = None,
        rollback_manager: AutomatedRollbackManager | None = None,
        reflection_settings: ReflectionTomlSettings | None = None,
        patch_event_sink: ReflectionPatchEventSink | None = None,
        orchestrator_agent_id: str = "",
    ) -> None:
        self._workspace_parent = workspace_parent
        self._team_workspace = (team_workspace or Path.cwd()).expanduser().resolve()
        self._project_root = project_root
        self._gate_runner: GateRunner = gate_runner or run_reliability_gate_subprocess
        self._rollback = rollback_manager or AutomatedRollbackManager()
        self._reflection_settings = reflection_settings
        self._patch_event_sink = patch_event_sink
        self._orchestrator_agent_id = orchestrator_agent_id

    def propose_patch(self, trace: list[TraceEvent]) -> ReflectionPatch | None:
        if not trace:
            return None

        settings = self._reflection_settings or load_reflection_toml_settings(self._team_workspace)

        root = self._project_root
        if root is None:
            root = resolve_project_root_for_gate()

        gate_result: ReliabilityGateResult | None = None
        passed = False
        if root is None:
            gate_result = ReliabilityGateResult(
                passed=False,
                returncode=-1,
                command=tuple(),
                stdout_tail="",
                stderr_tail=(
                    "Reliability Gate could not run: pass project_root= pointing at "
                    "a NAQSHA checkout that contains tests/."
                ),
            )
            passed = False
        else:
            gate_result = self._gate_runner(root)
            passed = gate_result.passed

        workspace = create_isolated_workspace(self._workspace_parent)

        effective_auto_merge = (
            settings.enabled
            and settings.auto_merge
            and passed
            and settings.reliability_gate
            and (self._team_workspace / "naqsha.toml").is_file()
        )

        md = build_candidate_markdown(trace, reliability_gate_passed=passed)
        (workspace / _CANDIDATE_MD).write_text(md, encoding="utf-8")

        auto_merged = False
        if effective_auto_merge:
            try:
                _write_merge_payload(self._team_workspace, workspace / _MERGE_DIR)
                patch_for_merge = ReflectionPatch(
                    workspace=workspace,
                    summary="",
                    reliability_gate_passed=True,
                    auto_merged=True,
                )
                self._rollback.merge_patch(
                    patch_for_merge,
                    team_workspace=self._team_workspace,
                    run_id=trace[0].run_id,
                    agent_id=self._orchestrator_agent_id,
                    event_sink=self._patch_event_sink,
                )
                auto_merged = True
            except (OSError, ValueError):
                auto_merged = False

        meta = build_meta_json(
            trace,
            reliability_gate_passed=passed,
            gate_result=gate_result,
            auto_merged=auto_merged,
        )
        (workspace / _META_JSON).write_text(meta, encoding="utf-8")

        if passed:
            msg = (
                "Reliability Gate passed. Human review is still required before any merge.\n"
                if not auto_merged
                else (
                    "Reliability Gate passed; patch was auto-merged into this team workspace. "
                    "Boot verification will run on the next agent run.\n"
                )
            )
            (workspace / _READINESS).write_text(msg, encoding="utf-8")
        else:
            (workspace / "GATE_FAILED.txt").write_text(
                "Reliability Gate did not pass. Do not treat this folder as review-ready.\n",
                encoding="utf-8",
            )

        notes = (
            "# Reviewed self-improvement\n\n"
            "- Read `CANDIDATE.md` and `meta.json` in this workspace.\n"
            "- Optional: save regression expectations with `naqsha eval save ...` "
            "into `.naqsha/evals/` and reference them when reviewing.\n"
        )
        if auto_merged:
            notes += (
                "- This patch was auto-merged per `[reflection] auto_merge = true`; "
                "the Automated Rollback Manager snapshots `naqsha.toml` and `tools/` "
                "before merge.\n"
            )
        else:
            notes += "- Do not merge or hotpatch the active runtime without human approval.\n"
        (workspace / _IMPROVEMENT_NOTES).write_text(notes, encoding="utf-8")

        short = (
            f"run {trace[0].run_id}: gate {'ok' if passed else 'failed'}; "
            f"see {workspace / _CANDIDATE_MD}"
        )
        if auto_merged:
            short += "; auto-merged into team workspace"

        return ReflectionPatch(
            workspace=workspace,
            summary=short,
            reliability_gate_passed=passed,
            auto_merged=auto_merged,
        )

propose_patch 

propose_patch(
    trace: list[TraceEvent],
) -> ReflectionPatch | None
Source code in src/naqsha/reflection/loop.py 
 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
def propose_patch(self, trace: list[TraceEvent]) -> ReflectionPatch | None:
    if not trace:
        return None

    settings = self._reflection_settings or load_reflection_toml_settings(self._team_workspace)

    root = self._project_root
    if root is None:
        root = resolve_project_root_for_gate()

    gate_result: ReliabilityGateResult | None = None
    passed = False
    if root is None:
        gate_result = ReliabilityGateResult(
            passed=False,
            returncode=-1,
            command=tuple(),
            stdout_tail="",
            stderr_tail=(
                "Reliability Gate could not run: pass project_root= pointing at "
                "a NAQSHA checkout that contains tests/."
            ),
        )
        passed = False
    else:
        gate_result = self._gate_runner(root)
        passed = gate_result.passed

    workspace = create_isolated_workspace(self._workspace_parent)

    effective_auto_merge = (
        settings.enabled
        and settings.auto_merge
        and passed
        and settings.reliability_gate
        and (self._team_workspace / "naqsha.toml").is_file()
    )

    md = build_candidate_markdown(trace, reliability_gate_passed=passed)
    (workspace / _CANDIDATE_MD).write_text(md, encoding="utf-8")

    auto_merged = False
    if effective_auto_merge:
        try:
            _write_merge_payload(self._team_workspace, workspace / _MERGE_DIR)
            patch_for_merge = ReflectionPatch(
                workspace=workspace,
                summary="",
                reliability_gate_passed=True,
                auto_merged=True,
            )
            self._rollback.merge_patch(
                patch_for_merge,
                team_workspace=self._team_workspace,
                run_id=trace[0].run_id,
                agent_id=self._orchestrator_agent_id,
                event_sink=self._patch_event_sink,
            )
            auto_merged = True
        except (OSError, ValueError):
            auto_merged = False

    meta = build_meta_json(
        trace,
        reliability_gate_passed=passed,
        gate_result=gate_result,
        auto_merged=auto_merged,
    )
    (workspace / _META_JSON).write_text(meta, encoding="utf-8")

    if passed:
        msg = (
            "Reliability Gate passed. Human review is still required before any merge.\n"
            if not auto_merged
            else (
                "Reliability Gate passed; patch was auto-merged into this team workspace. "
                "Boot verification will run on the next agent run.\n"
            )
        )
        (workspace / _READINESS).write_text(msg, encoding="utf-8")
    else:
        (workspace / "GATE_FAILED.txt").write_text(
            "Reliability Gate did not pass. Do not treat this folder as review-ready.\n",
            encoding="utf-8",
        )

    notes = (
        "# Reviewed self-improvement\n\n"
        "- Read `CANDIDATE.md` and `meta.json` in this workspace.\n"
        "- Optional: save regression expectations with `naqsha eval save ...` "
        "into `.naqsha/evals/` and reference them when reviewing.\n"
    )
    if auto_merged:
        notes += (
            "- This patch was auto-merged per `[reflection] auto_merge = true`; "
            "the Automated Rollback Manager snapshots `naqsha.toml` and `tools/` "
            "before merge.\n"
        )
    else:
        notes += "- Do not merge or hotpatch the active runtime without human approval.\n"
    (workspace / _IMPROVEMENT_NOTES).write_text(notes, encoding="utf-8")

    short = (
        f"run {trace[0].run_id}: gate {'ok' if passed else 'failed'}; "
        f"see {workspace / _CANDIDATE_MD}"
    )
    if auto_merged:
        short += "; auto-merged into team workspace"

    return ReflectionPatch(
        workspace=workspace,
        summary=short,
        reliability_gate_passed=passed,
        auto_merged=auto_merged,
    )

ReliabilityGateResult dataclass

Source code in src/naqsha/reflection/reliability_gate.py 
21
22
23
24
25
26
27
@dataclass(frozen=True)
class ReliabilityGateResult:
    passed: bool
    returncode: int
    command: tuple[str, ...]
    stdout_tail: str
    stderr_tail: str

passed instance-attribute 

passed: bool

returncode instance-attribute 

returncode: int

command instance-attribute 

command: tuple[str, ...]

stdout_tail instance-attribute 

stdout_tail: str

stderr_tail instance-attribute 

stderr_tail: str

AutomatedRollbackManager

Snapshot naqsha.toml and tools/ before merge; restore on failed boot.

Source code in src/naqsha/reflection/rollback.py 
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
227
228
229
230
231
232
233
234
235
236
237
class AutomatedRollbackManager:
    """Snapshot ``naqsha.toml`` and ``tools/`` before merge; restore on failed boot."""

    def merge_patch(
        self,
        patch: ReflectionPatch,
        *,
        team_workspace: Path,
        run_id: str,
        agent_id: str = "",
        event_sink: ReflectionPatchEventSink | None = None,
    ) -> None:
        """Backup tracked files, apply ``patch.workspace/{merge}/``, set boot pending."""

        merge_src = patch.workspace / _MERGE_DIR_NAME
        if not merge_src.is_dir():
            msg = f"Patch workspace missing {_MERGE_DIR_NAME}/ directory for merge."
            raise ValueError(msg)

        team_workspace = team_workspace.expanduser().resolve()
        n = _naqsha_dir(team_workspace)
        n.mkdir(parents=True, exist_ok=True)
        backups_root = n / BACKUPS_SUBDIR
        backups_root.mkdir(parents=True, exist_ok=True)

        stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S_%f")
        backup_dir = backups_root / stamp
        _backup_merge_targets(team_workspace, backup_dir)

        _apply_merge_tree(merge_src, team_workspace)

        patch_id = patch.workspace.name
        meta: dict[str, Any] = {
            "patch_id": patch_id,
            "run_id": run_id,
            "agent_id": agent_id,
            "backup_name": backup_dir.name,
        }
        meta_text = json.dumps(meta, indent=2, sort_keys=True)
        (n / LAST_MERGE_META).write_text(meta_text, encoding="utf-8")
        _write_boot_status(team_workspace, STATUS_PENDING)
        _prune_backups(backups_root)

        if event_sink is not None:
            event_sink.patch_merged(
                run_id=run_id,
                agent_id=agent_id,
                patch_id=patch_id,
                auto_merged=patch.auto_merged,
            )

    def verify_boot_if_pending(
        self,
        team_workspace: Path,
        *,
        health_check: Callable[[], bool],
        event_sink: ReflectionPatchEventSink | None = None,
    ) -> bool:
        """If boot status is ``pending``, run *health_check*; rollback on failure."""

        team_workspace = team_workspace.expanduser().resolve()
        status = _read_boot_status(team_workspace)

        if status == STATUS_ROLLED_BACK:
            if health_check():
                _write_boot_status(team_workspace, STATUS_STABLE)
            return True

        if status != STATUS_PENDING:
            return True

        if health_check():
            _write_boot_status(team_workspace, STATUS_STABLE)
            return True

        n = _naqsha_dir(team_workspace)
        meta_path = n / LAST_MERGE_META
        backup_name: str | None = None
        patch_id = ""
        run_id = ""
        agent_id = ""
        if meta_path.is_file():
            try:
                meta = json.loads(meta_path.read_text(encoding="utf-8"))
                backup_name = meta.get("backup_name") if isinstance(meta, dict) else None
                patch_id = str(meta.get("patch_id", "")) if isinstance(meta, dict) else ""
                run_id = str(meta.get("run_id", "")) if isinstance(meta, dict) else ""
                agent_id = str(meta.get("agent_id", "")) if isinstance(meta, dict) else ""
            except json.JSONDecodeError:
                backup_name = None

        backups_root = n / BACKUPS_SUBDIR
        backup_dir: Path | None = None
        if backup_name and (backups_root / backup_name).is_dir():
            backup_dir = backups_root / backup_name
        else:
            dirs = sorted(
                (p for p in backups_root.iterdir() if p.is_dir()),
                key=lambda p: p.name,
                reverse=True,
            )
            if len(dirs) == 1:
                backup_dir = dirs[0]
            else:
                backup_dir = None

        reason = "Boot health check failed after auto-merge."
        if backup_dir is not None:
            _restore_merge_targets(team_workspace, backup_dir)
        else:
            reason += (
                " No unambiguous backup (need valid last_merge_meta or exactly one backup); "
                "workspace left unchanged."
            )

        _write_boot_status(team_workspace, STATUS_ROLLED_BACK)

        if event_sink is not None:
            event_sink.patch_rolled_back(
                run_id=run_id,
                agent_id=agent_id,
                patch_id=patch_id or "unknown",
                reason=reason,
            )

        return True

merge_patch 

merge_patch(
    patch: ReflectionPatch,
    *,
    team_workspace: Path,
    run_id: str,
    agent_id: str = "",
    event_sink: ReflectionPatchEventSink | None = None,
) -> None

Backup tracked files, apply patch.workspace/{merge}/, set boot pending.

Source code in src/naqsha/reflection/rollback.py 
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
def merge_patch(
    self,
    patch: ReflectionPatch,
    *,
    team_workspace: Path,
    run_id: str,
    agent_id: str = "",
    event_sink: ReflectionPatchEventSink | None = None,
) -> None:
    """Backup tracked files, apply ``patch.workspace/{merge}/``, set boot pending."""

    merge_src = patch.workspace / _MERGE_DIR_NAME
    if not merge_src.is_dir():
        msg = f"Patch workspace missing {_MERGE_DIR_NAME}/ directory for merge."
        raise ValueError(msg)

    team_workspace = team_workspace.expanduser().resolve()
    n = _naqsha_dir(team_workspace)
    n.mkdir(parents=True, exist_ok=True)
    backups_root = n / BACKUPS_SUBDIR
    backups_root.mkdir(parents=True, exist_ok=True)

    stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S_%f")
    backup_dir = backups_root / stamp
    _backup_merge_targets(team_workspace, backup_dir)

    _apply_merge_tree(merge_src, team_workspace)

    patch_id = patch.workspace.name
    meta: dict[str, Any] = {
        "patch_id": patch_id,
        "run_id": run_id,
        "agent_id": agent_id,
        "backup_name": backup_dir.name,
    }
    meta_text = json.dumps(meta, indent=2, sort_keys=True)
    (n / LAST_MERGE_META).write_text(meta_text, encoding="utf-8")
    _write_boot_status(team_workspace, STATUS_PENDING)
    _prune_backups(backups_root)

    if event_sink is not None:
        event_sink.patch_merged(
            run_id=run_id,
            agent_id=agent_id,
            patch_id=patch_id,
            auto_merged=patch.auto_merged,
        )

verify_boot_if_pending 

verify_boot_if_pending(
    team_workspace: Path,
    *,
    health_check: Callable[[], bool],
    event_sink: ReflectionPatchEventSink | None = None,
) -> bool

If boot status is pending, run health_check; rollback on failure.

Source code in src/naqsha/reflection/rollback.py 
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
227
228
229
230
231
232
233
234
235
236
237
def verify_boot_if_pending(
    self,
    team_workspace: Path,
    *,
    health_check: Callable[[], bool],
    event_sink: ReflectionPatchEventSink | None = None,
) -> bool:
    """If boot status is ``pending``, run *health_check*; rollback on failure."""

    team_workspace = team_workspace.expanduser().resolve()
    status = _read_boot_status(team_workspace)

    if status == STATUS_ROLLED_BACK:
        if health_check():
            _write_boot_status(team_workspace, STATUS_STABLE)
        return True

    if status != STATUS_PENDING:
        return True

    if health_check():
        _write_boot_status(team_workspace, STATUS_STABLE)
        return True

    n = _naqsha_dir(team_workspace)
    meta_path = n / LAST_MERGE_META
    backup_name: str | None = None
    patch_id = ""
    run_id = ""
    agent_id = ""
    if meta_path.is_file():
        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))
            backup_name = meta.get("backup_name") if isinstance(meta, dict) else None
            patch_id = str(meta.get("patch_id", "")) if isinstance(meta, dict) else ""
            run_id = str(meta.get("run_id", "")) if isinstance(meta, dict) else ""
            agent_id = str(meta.get("agent_id", "")) if isinstance(meta, dict) else ""
        except json.JSONDecodeError:
            backup_name = None

    backups_root = n / BACKUPS_SUBDIR
    backup_dir: Path | None = None
    if backup_name and (backups_root / backup_name).is_dir():
        backup_dir = backups_root / backup_name
    else:
        dirs = sorted(
            (p for p in backups_root.iterdir() if p.is_dir()),
            key=lambda p: p.name,
            reverse=True,
        )
        if len(dirs) == 1:
            backup_dir = dirs[0]
        else:
            backup_dir = None

    reason = "Boot health check failed after auto-merge."
    if backup_dir is not None:
        _restore_merge_targets(team_workspace, backup_dir)
    else:
        reason += (
            " No unambiguous backup (need valid last_merge_meta or exactly one backup); "
            "workspace left unchanged."
        )

    _write_boot_status(team_workspace, STATUS_ROLLED_BACK)

    if event_sink is not None:
        event_sink.patch_rolled_back(
            run_id=run_id,
            agent_id=agent_id,
            patch_id=patch_id or "unknown",
            reason=reason,
        )

    return True

ReflectionWorkspaceError

Bases: ValueError

Reflection would write inside the naqsha package tree or another forbidden path.

Source code in src/naqsha/reflection/workspace.py 
11
12
class ReflectionWorkspaceError(ValueError):
    """Reflection would write inside the naqsha package tree or another forbidden path."""

build_candidate_markdown 

build_candidate_markdown(
    events: list[TraceEvent],
    *,
    reliability_gate_passed: bool,
) -> str
Source code in src/naqsha/reflection/candidate.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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
def build_candidate_markdown(events: list[TraceEvent], *, reliability_gate_passed: bool) -> str:
    if not events:
        return "# Reflection candidate\n\n(empty trace)\n"

    run_id = events[0].run_id
    lines: list[str] = [
        f"# Reflection candidate for run `{run_id}`",
        "",
        f"- generated_at: `{datetime.now(tz=UTC).isoformat()}`",
        f"- reliability_gate_passed: `{reliability_gate_passed}`",
        f"- ready_for_human_review: `{reliability_gate_passed}`",
        "",
        "## Trace outcomes",
        "",
    ]

    failures = [e for e in events if e.kind == "failure"]
    answer = next((e.payload["answer"] for e in reversed(events) if e.kind == "answer"), None)
    if failures:
        lines.append("Failure events:")
        for f in failures:
            code = f.payload.get("code", "?")
            msg = f.payload.get("message", "")
            lines.append(f"- `{code}`: {msg}")
        lines.append("")
    else:
        lines.append("No failure events recorded.")
        lines.append("")

    if answer is not None:
        lines.extend(["## Final answer", "", answer, ""])
    else:
        lines.extend(["## Final answer", "", "(none)", ""])

    chron = tool_calls_chronology(events)
    lines.extend(["## Tool call path (chronology)", ""])
    for row in chron:
        lines.append(f"- `{row['tool']}` call_id `{row['call_id']}`")
    if not chron:
        lines.append("(no tool calls in trace actions)")
    lines.append("")

    lines.extend(
        [
            "## Reviewed self-improvement checklist (human only)",
            "",
            "- Consider capturing a regression fixture: `naqsha eval save RUN_ID <name>` then "
            "`naqsha eval check RUN_ID --name <name>`.",
            "- Prefer Run Profile changes (model, budgets, allowed tools, memory adapter) over "
            "prompt-only guidance.",
            "- If tool selection was wrong, tighten tool descriptions/schemas in code and add "
            "deterministic replay tests before widening **Tool Policy**.",
            "- Do not merge **Reflection Patch** workspaces into the runtime without review; "
            "no hotpatching **Approval Gate** or policy from trace text.",
            "",
        ]
    )
    lines.extend(
        [
            "## Guidance (untrusted; for humans only)",
            "",
            "This file summarizes observable trace facts. It does not authorize runtime, "
            "policy, or approval changes.",
            "",
            "- If failures reference budgets or policy denials, inspect the trace and "
            "red-team fixtures before changing Tool Policy.",
            "- Prefer targeted test updates over widening approvals or bypassing gates.",
            "",
        ]
    )
    return "\n".join(lines)

build_meta_json 

build_meta_json(
    events: list[TraceEvent],
    *,
    reliability_gate_passed: bool,
    gate_result: ReliabilityGateResult | None,
    auto_merged: bool = False,
) -> str
Source code in src/naqsha/reflection/candidate.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def build_meta_json(
    events: list[TraceEvent],
    *,
    reliability_gate_passed: bool,
    gate_result: ReliabilityGateResult | None,
    auto_merged: bool = False,
) -> str:
    run_id = events[0].run_id if events else ""
    payload: dict[str, object] = {
        "run_id": run_id,
        "reliability_gate_passed": reliability_gate_passed,
        "ready_for_human_review": reliability_gate_passed,
        "auto_merged": auto_merged,
    }
    if gate_result is not None:
        payload["reliability_gate"] = {
            "passed": gate_result.passed,
            "returncode": gate_result.returncode,
            "command": list(gate_result.command),
        }
    return json.dumps(payload, indent=2, sort_keys=True)

load_reflection_toml_settings 

load_reflection_toml_settings(
    team_workspace: Path,
) -> ReflectionTomlSettings

Parse [reflection] from team_workspace/naqsha.toml if present.

Source code in src/naqsha/reflection/config.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def load_reflection_toml_settings(team_workspace: Path) -> ReflectionTomlSettings:
    """Parse ``[reflection]`` from ``team_workspace/naqsha.toml`` if present."""

    path = team_workspace / "naqsha.toml"
    if not path.is_file():
        return ReflectionTomlSettings()
    with path.open("rb") as fh:
        data = tomllib.load(fh)
    blob = data.get("reflection")
    if not isinstance(blob, dict):
        return ReflectionTomlSettings()
    return ReflectionTomlSettings(
        enabled=_as_bool(blob.get("enabled", False), False),
        auto_merge=_as_bool(blob.get("auto_merge", False), False),
        reliability_gate=_as_bool(blob.get("reliability_gate", True), True),
    )

failing_gate_runner 

failing_gate_runner(
    _project_root: Path,
) -> ReliabilityGateResult

Test helper: gate always fails.

Source code in src/naqsha/reflection/loop.py 
195
196
197
198
199
200
201
202
203
204
def failing_gate_runner(_project_root: Path) -> ReliabilityGateResult:
    """Test helper: gate always fails."""

    return ReliabilityGateResult(
        passed=False,
        returncode=1,
        command=("noop-fail",),
        stdout_tail="",
        stderr_tail="synthetic failure",
    )

noop_gate_runner 

noop_gate_runner(
    _project_root: Path,
) -> ReliabilityGateResult

Test helper: gate always passes.

Source code in src/naqsha/reflection/loop.py 
183
184
185
186
187
188
189
190
191
192
def noop_gate_runner(_project_root: Path) -> ReliabilityGateResult:
    """Test helper: gate always passes."""

    return ReliabilityGateResult(
        passed=True,
        returncode=0,
        command=("noop",),
        stdout_tail="",
        stderr_tail="",
    )

resolve_project_root_for_gate 

resolve_project_root_for_gate() -> Path | None

Find a development checkout root that contains the gate test files.

Source code in src/naqsha/reflection/reliability_gate.py 
30
31
32
33
34
35
36
37
38
def resolve_project_root_for_gate() -> Path | None:
    """Find a development checkout root that contains the gate test files."""

    here = Path(__file__).resolve().parent
    for base in [here] + list(here.parents):
        relay = base / RELIABILITY_GATE_TEST_PATHS[0]
        if relay.is_file():
            return base
    return None

run_reliability_gate_subprocess 

run_reliability_gate_subprocess(
    project_root: Path,
) -> ReliabilityGateResult

Run pytest on the Reliability Gate corpus (blocking).

Source code in src/naqsha/reflection/reliability_gate.py 
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
def run_reliability_gate_subprocess(project_root: Path) -> ReliabilityGateResult:
    """Run pytest on the Reliability Gate corpus (blocking)."""

    root = project_root.resolve()
    missing = [p for p in RELIABILITY_GATE_TEST_PATHS if not (root / p).is_file()]
    if missing:
        return ReliabilityGateResult(
            passed=False,
            returncode=-1,
            command=tuple(),
            stdout_tail="",
            stderr_tail=(
                "Reliability Gate tests missing under project root "
                f"{root}: {', '.join(missing)}"
            ),
        )

    argv = [
        sys.executable,
        "-m",
        "pytest",
        "-q",
        *[str(root / p) for p in RELIABILITY_GATE_TEST_PATHS],
    ]
    proc = subprocess.run(
        argv,
        cwd=root,
        capture_output=True,
        text=True,
        timeout=900,
        check=False,
    )
    return ReliabilityGateResult(
        passed=proc.returncode == 0,
        returncode=proc.returncode,
        command=tuple(argv),
        stdout_tail=_truncate(proc.stdout or ""),
        stderr_tail=_truncate(proc.stderr or ""),
    )

assert_workspace_outside_package 

assert_workspace_outside_package(
    path: Path, *, label: str = "path"
) -> None

Ensure path is not inside the installed naqsha package directory.

Source code in src/naqsha/reflection/workspace.py 
19
20
21
22
23
24
25
26
27
28
29
30
def assert_workspace_outside_package(path: Path, *, label: str = "path") -> None:
    """Ensure ``path`` is not inside the installed ``naqsha`` package directory."""

    pkg = naqsha_package_dir()
    resolved = path.expanduser().resolve()
    try:
        resolved.relative_to(pkg)
    except ValueError:
        return
    raise ReflectionWorkspaceError(
        f"{label} {resolved} must not lie inside the naqsha package directory {pkg}."
    )

create_isolated_workspace 

create_isolated_workspace(
    parent: Path, *, prefix: str = "reflection-patch"
) -> Path

Create parent / {prefix}-{uuid} after validating parent.

Source code in src/naqsha/reflection/workspace.py 
33
34
35
36
37
38
39
40
41
def create_isolated_workspace(parent: Path, *, prefix: str = "reflection-patch") -> Path:
    """Create ``parent / {prefix}-{uuid}`` after validating ``parent``."""

    assert_workspace_outside_package(parent, label="workspace_parent")
    parent.mkdir(parents=True, exist_ok=True)
    sub = parent / f"{prefix}-{uuid.uuid4().hex[:10]}"
    sub.mkdir(parents=False, exist_ok=False)
    assert_workspace_outside_package(sub, label="workspace")
    return sub

naqsha_package_dir 

naqsha_package_dir() -> Path
Source code in src/naqsha/reflection/workspace.py 
15
16
def naqsha_package_dir() -> Path:
    return Path(naqsha.__file__).resolve().parent