Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 113 additions & 0 deletions bin/_layer5.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
"""bin/_layer5.py — Layer 5 self-interrogation trace helper (v1.3 prototype).

Provides build_trace_record() used by walker and eval_metadata at the three
named choice points:
- walker-concern : concern-resolution in walker.record_answer
- substitution : substitution logged in eval_metadata.write_sidecar
- exemplar-binding : exemplar binding committed in walker

Schema (six fields + timestamp, machine-parseable JSON):
{
"choice_point": "walker-concern" | "substitution" | "exemplar-binding",
"step_or_concern_id": "<id>",
"options_considered": ["<a>", "<b>", "<c>"],
"selected": "<a>",
"rationale": "<one-or-two-sentence-why>",
"validation_anchor": "<tier-or-check | null>",
"source_anchor": "<spec-section-or-ADR-id | null>",
"timestamp": "<ISO8601>"
}

Emission is skipped when:
- SPECTRE_LAYER5=off environment variable is set
- len(options_considered) <= 1 (no real choice)

Stdlib only. No third-party dependencies.
"""
from __future__ import annotations

import os
from datetime import datetime, timezone
from typing import Any

# Valid choice-point names.
CHOICE_POINTS: frozenset[str] = frozenset({
"walker-concern",
"substitution",
"exemplar-binding",
})

_LAYER5_DISABLED_SENTINEL = "off"


def _layer5_enabled() -> bool:
"""Return False when SPECTRE_LAYER5=off disables trace emission."""
return os.environ.get("SPECTRE_LAYER5", "").lower() != _LAYER5_DISABLED_SENTINEL


def build_trace_record(
*,
choice_point: str,
step_or_concern_id: str,
options_considered: list[str],
selected: str,
rationale: str,
validation_anchor: str | None,
source_anchor: str | None,
) -> dict[str, Any] | None:
"""Build a Layer 5 trace record dict and return it, or None if emission is skipped.

Returns None (skips emission) when:
- SPECTRE_LAYER5=off
- len(options_considered) <= 1 (no real choice to trace)

The caller appends the returned dict to state.layer5_trace (walker) or
includes it in the sidecar payload projection (eval_metadata).
"""
if not _layer5_enabled():
return None
if len(options_considered) <= 1:
return None
if choice_point not in CHOICE_POINTS:
raise ValueError(
f"unknown choice_point {choice_point!r}; expected one of {sorted(CHOICE_POINTS)}"
)
return {
"choice_point": choice_point,
"step_or_concern_id": step_or_concern_id,
"options_considered": list(options_considered),
"selected": selected,
"rationale": rationale,
"validation_anchor": validation_anchor,
"source_anchor": source_anchor,
"timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
}


def project_substitution_trace(substitution: dict[str, Any]) -> dict[str, Any] | None:
"""Project a substitution dict into a layer5_trace record.

The substitution dict (v1.2.1 #6 schema extended by v1.3 #10) may carry
optional layer5 fields:
options_considered, selected, rationale, validation_anchor, source_anchor

Returns a trace record if enough fields are present and emission is not
disabled, else None.
"""
options = substitution.get("options_considered")
if not isinstance(options, list):
return None
selected = substitution.get("selected", substitution.get("to", ""))
rationale = substitution.get("rationale", substitution.get("reason", ""))
step_id = substitution.get("step_id", "")
validation_anchor = substitution.get("validation_anchor")
source_anchor = substitution.get("source_anchor")
return build_trace_record(
choice_point="substitution",
step_or_concern_id=step_id,
options_considered=options,
selected=selected,
rationale=rationale,
validation_anchor=validation_anchor,
source_anchor=source_anchor,
)
15 changes: 15 additions & 0 deletions bin/eval_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,6 +241,7 @@ def write_sidecar(
substrate_resolution: dict | None = None,
findings_inline: list[dict] | None = None,
substitutions: list[dict] | None = None,
layer5_trace: list[dict] | None = None,
) -> pathlib.Path:
"""Atomic write of <spec>.eval.json next to the spec file.

Expand Down Expand Up @@ -275,6 +276,12 @@ def write_sidecar(
"tier1_check_name": <kind>, "step_id": <step-N>}`` recording an agent's
rewrite of action content or verification commands to satisfy a Tier-1
check. Not a finding — contemporaneous evidence the operator can audit.

*layer5_trace* — v1.3 #10 reasoning-trace records. When non-None, written
verbatim under the ``layer5_trace`` key. Each entry is a six-field dict
(choice_point, step_or_concern_id, options_considered, selected, rationale,
validation_anchor, source_anchor, timestamp). Projected from walker state
or from substitution dicts that carry the extended schema.
"""
spec_path = pathlib.Path(spec_path)
sidecar_path = sidecar_path_for(spec_path)
Expand Down Expand Up @@ -323,6 +330,13 @@ def write_sidecar(
if substitutions is not None:
payload["substitutions"] = substitutions

# v1.3 #10: Layer 5 reasoning-trace records. Each entry is a six-field
# dict (choice_point, step_or_concern_id, options_considered, selected,
# rationale, validation_anchor, source_anchor, timestamp). Projected
# from walker state or from substitution dicts with extended schema.
if layer5_trace is not None:
payload["layer5_trace"] = layer5_trace

# Atomic write: mkstemp + os.replace
fd, tmp = tempfile.mkstemp(
dir=sidecar_path.parent, prefix=sidecar_path.name, suffix=".tmp"
Expand Down Expand Up @@ -506,6 +520,7 @@ def write_envelope_alongside_sidecar(
substrate_resolution=payload.get("substrate_resolution"),
findings_inline=payload.get("findings_inline"),
substitutions=payload.get("substitutions"),
layer5_trace=payload.get("layer5_trace"),
)
except KeyError as exc:
_status.emit("error", "eval_metadata.sidecar_missing_field", dest="stderr",
Expand Down
49 changes: 49 additions & 0 deletions bin/walker.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
sys.path.insert(0, str(_ROOT))

from bin import _catalog # noqa: E402
from bin import _layer5 # noqa: E402

WALKER_VERSION = "1.0.0"

Expand Down Expand Up @@ -94,6 +95,8 @@ class WalkState:
human_user_asked: bool = False
integrator_asked: bool = False
operator_asked: bool = False
# v1.3 #10 — Layer 5 reasoning-trace records accumulated during this walk.
layer5_trace: list[dict] = field(default_factory=list)


_OQ_INLINE_RE = re.compile(
Expand Down Expand Up @@ -312,6 +315,32 @@ def record_answer(state: WalkState, *, concern_id: str, answer: str) -> WalkStat
for i, c in enumerate(state.pending):
if c.id == concern_id:
state.asked.append(c)
# v1.3 #10 — emit layer5 trace when a real choice exists.
# Skip when prefab_options has 0 or 1 entry (no meaningful choice).
if len(c.prefab_options) > 1:
_is_exemplar = concern_id in _EXEMPLAR_BINDING_CONCERN_IDS
_choice_point = "exemplar-binding" if _is_exemplar else "walker-concern"
_rationale = (
f"Operator bound exemplar '{answer}' for view concern '{concern_id}' "
f"from {len(c.prefab_options)} catalog options."
if _is_exemplar else
f"Operator selected '{answer}' for concern '{concern_id}' "
f"({c.summary[:80]}…) after considering "
f"{len(c.prefab_options)} prefab options."
)
_trace = _layer5.build_trace_record(
choice_point=_choice_point,
step_or_concern_id=concern_id,
options_considered=list(c.prefab_options),
selected=answer,
rationale=_rationale,
validation_anchor=(
"tier2-cross-view-gate" if _is_exemplar else "tier1-structural"
),
source_anchor=None,
)
if _trace is not None:
state.layer5_trace.append(_trace)
state.answered[concern_id] = answer
del state.pending[i]
state.round_count += 1
Expand Down Expand Up @@ -507,6 +536,8 @@ def persist(state: WalkState, path: pathlib.Path) -> None:
"human_user_asked": state.human_user_asked,
"integrator_asked": state.integrator_asked,
"operator_asked": state.operator_asked,
# v1.3 #10 — layer5 reasoning-trace records
"layer5_trace": list(state.layer5_trace),
}
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=path.name, suffix=".tmp")
Expand Down Expand Up @@ -569,6 +600,8 @@ def load(path: pathlib.Path) -> WalkState | None:
human_user_asked=data.get("human_user_asked", False),
integrator_asked=data.get("integrator_asked", False),
operator_asked=data.get("operator_asked", False),
# v1.3 #10 — layer5 trace (defaults to empty list for backwards compat)
layer5_trace=list(data.get("layer5_trace", [])),
)


Expand Down Expand Up @@ -918,6 +951,18 @@ def generate_semantic_criteria_concern(
# _VIEW_SCOPE_CONCERN_IDS, kept alongside for co-location).
_SCOPE_CONCERN_TO_VIEW: dict[str, str] = {v: k for k, v in _VIEW_SCOPE_CONCERN_IDS.items()}

# v1.3 #10 — Concern IDs that represent exemplar-binding choices (catalog slug
# selection). When one of these is answered via record_answer, a layer5 trace
# with choice_point="exemplar-binding" is emitted instead of "walker-concern".
_EXEMPLAR_BINDING_CONCERN_IDS: frozenset[str] = frozenset({
"input-exemplar-pi",
"help-text-style-hu",
"error-text-style-hu",
"api-exemplar-int",
"log-format-style-op",
"observability-style-op",
})


def _exemplar_options_for(
fingerprint: str | None,
Expand Down Expand Up @@ -2416,6 +2461,8 @@ def _drive_to_completeness_satisfied(
"human_user_asked": state.human_user_asked,
"integrator_asked": state.integrator_asked,
"operator_asked": state.operator_asked,
# v1.3 #10 — layer5 reasoning-trace records
"layer5_trace": list(state.layer5_trace),
}
print(json.dumps(payload, indent=2, sort_keys=True))
else:
Expand Down Expand Up @@ -2591,6 +2638,8 @@ def _drive_to_completeness_satisfied(
"human_user_asked": state.human_user_asked,
"integrator_asked": state.integrator_asked,
"operator_asked": state.operator_asked,
# v1.3 #10 — layer5 reasoning-trace records
"layer5_trace": list(state.layer5_trace),
}
print(json.dumps(payload, indent=2, sort_keys=True))
else:
Expand Down
14 changes: 14 additions & 0 deletions docs/glossary.md
Original file line number Diff line number Diff line change
Expand Up @@ -1234,3 +1234,17 @@ Status codes: dotted identifiers like `walker.init`. Terms: `term:<noun>` prefix
- user_action: No action required. Monitor round and pending counts to gauge walk progress. Operator interpretation only — walker.round does not imply any threshold or convergence signal.
- related: walker.yield, walker.coverage
- since: v1.2

## term:layer5-trace
- kind: term
- dev: A Layer 5 reasoning-trace record emitted at a named choice point (walker-concern, substitution, or exemplar-binding). Six fields: choice_point, step_or_concern_id, options_considered, selected, rationale, validation_anchor, source_anchor, plus timestamp. Written to the `.eval.json` sidecar under `layer5_trace: [...]`. Skipped when SPECTRE_LAYER5=off or when options_considered has <=1 entry (no real choice).
- pm: A logged record of why the system made a specific choice at a key decision point. This audit trail lets reviewers verify the reasoning behind each significant design or binding selection.
- related: term:layer5-choice-point, eval.substitutions
- since: v1.3

## term:layer5-choice-point
- kind: term
- dev: One of three named decision sites where Layer 5 reasoning-trace records are emitted. (1) walker-concern — operator selects from prefab options when answering a walker concern. (2) substitution — agent rewrites action or verification content to satisfy a Tier-1 check. (3) exemplar-binding — operator selects a catalog exemplar for a view. Traces are emitted only when a real choice exists (options_considered > 1).
- pm: The three types of decisions the system records reasoning traces for: answering a walker question, rewriting content to pass a check, or picking a catalog example to follow.
- related: term:layer5-trace
- since: v1.3
Loading
Loading