Tighten control-plane subsystem lookup context

This commit is contained in:
rcourtman
2026-03-30 09:42:16 +01:00
parent 444e827392
commit 40e095a9a0
12 changed files with 380 additions and 25 deletions
+1 -1
View File
@@ -18,7 +18,7 @@ Use:
7. `python3 scripts/release_control/contract_audit.py --check` if you need a machine-derived subsystem contract audit, including explicit cross-subsystem dependency checks and exact registry-derived shared-boundary wording
Local pre-commit runs the v6 machine audits against staged control-file content so partial staging cannot hide governance drift.
Local pre-commit also blocks partial staging for hook-sensitive governance files under `docs/release-control/v6/`, `scripts/release_control/`, `internal/repoctl/`, `.husky/pre-commit`, and `.github/workflows/canonical-governance.yml`, because those checks still execute or structurally read the working-tree versions locally.
8. `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...]` if you need subsystem ownership, proof routing, lane context, relevant decision records, and dependent contract-update obligations for a change
8. `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty --lean` if you need subsystem ownership, proof routing, exact contract-focus lines, and compact lane context for a change
For governed runtime changes, a staged subsystem contract only counts if its
diff updates a substantive contract section such as `Purpose`, `Canonical Files`,
+2 -1
View File
@@ -81,7 +81,8 @@ Useful helper tools:
- `python3 scripts/release_control/readiness_assertion_guard.py --active-target --proof-type hybrid`
Runs any executable hybrid proof commands required by the active target phase, without replacing the linked manual release gates.
- `python3 scripts/release_control/registry_audit.py --check`
- `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...]` for subsystem ownership, proof routing, lane context, relevant decision records, and dependent contract-update obligations
- `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty --lean` for subsystem ownership, proof routing, exact contract-focus lines, and compact lane context without pulling the full governance payload into every lookup
Omit `--lean` when you explicitly need the full decision, release-gate, and status-summary payload.
- `python3 scripts/release_control/work_claim.py --kind <KIND> --id <ID> --summary <SUMMARY> --agent-id <AGENT_ID> --pretty`
Reserve or renew exactly one governed slice in a shared checkout before mutation without hand-editing `status.json`.
- `python3 scripts/release_control/worktree_base.py --base-branch <BASE_BRANCH> --pretty`
@@ -482,10 +482,11 @@ Before adding new behavior, the author must answer:
If those answers are not obvious in under a minute, the subsystem still needs
architectural hardening.
Use `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...]`
to ask the repo which subsystem, contract, proof route, live lane context,
relevant decision records, and dependent contract-update obligations apply to a
file set before editing.
Use `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty --lean`
to ask the repo which subsystem, contract, proof route, exact contract-focus
lines, and compact lane context apply to a file set before editing.
Omit `--lean` only when the full decision, release-gate, or status-summary
payload is materially needed.
## Boundary Rule
+1 -1
View File
@@ -18,7 +18,7 @@ Use:
7. `python3 scripts/release_control/contract_audit.py --check` if you need a machine-derived subsystem contract audit, including explicit cross-subsystem dependency checks and exact registry-derived shared-boundary wording
Local pre-commit runs the v6 machine audits against staged control-file content so partial staging cannot hide governance drift.
Local pre-commit also blocks partial staging for hook-sensitive governance files under `docs/release-control/v6/`, `scripts/release_control/`, `internal/repoctl/`, `.husky/pre-commit`, and `.github/workflows/canonical-governance.yml`, because those checks still execute or structurally read the working-tree versions locally.
8. `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...]` if you need subsystem ownership, proof routing, lane context, relevant decision records, and dependent contract-update obligations for a change
8. `python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty --lean` if you need subsystem ownership, proof routing, exact contract-focus lines, and compact lane context for a change
For governed runtime changes, a staged subsystem contract only counts if its
diff updates a substantive contract section such as `Purpose`, `Canonical Files`,
@@ -303,10 +303,31 @@ def required_contract_updates(
descriptor = f"{reference['heading']}: {reference['path']}"
if descriptor not in entry["matched_references"]:
entry["matched_references"].append(descriptor)
detail = {
"heading": str(reference.get("heading", "")),
"path": str(reference.get("path", "")),
}
if isinstance(reference.get("line"), int):
detail["line"] = int(reference["line"])
if isinstance(reference.get("heading_line"), int):
detail["heading_line"] = int(reference["heading_line"])
detail_list = entry.setdefault("matched_reference_details", [])
if detail not in detail_list:
detail_list.append(detail)
for data in required.values():
data["touched_runtime_files"] = sorted(set(data["touched_runtime_files"]))
data["matched_references"] = sorted(set(data["matched_references"]), key=str.casefold)
details = data.get("matched_reference_details")
if isinstance(details, list) and details:
data["matched_reference_details"] = sorted(
details,
key=lambda item: (
int(item.get("line", 0) or 0),
str(item.get("heading", "")).casefold(),
str(item.get("path", "")).casefold(),
),
)
return dict(sorted(required.items()))
@@ -2708,6 +2708,23 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"## Extension Points: internal/unifiedresources/views.go",
],
)
self.assertEqual(
required["docs/release-control/v6/internal/subsystems/monitoring.md"]["matched_reference_details"],
[
{
"heading": "## Canonical Files",
"path": "internal/unifiedresources/views.go",
"line": 32,
"heading_line": 23,
},
{
"heading": "## Extension Points",
"path": "internal/unifiedresources/views.go",
"line": 46,
"heading_line": 42,
},
],
)
def test_monitoring_owned_runtime_does_not_require_unified_resources_contract(self):
required = required_contract_updates(["internal/monitoring/monitor.go"])
+1 -1
View File
@@ -334,7 +334,7 @@ def agent_entrypoint(*, staged: bool = False) -> dict[str, Any]:
"python3 scripts/release_control/status_lookup.py --followup <FOLLOWUP_ID> --pretty",
"python3 scripts/release_control/status_lookup.py --work-claim <CLAIM_ID> --pretty",
"python3 scripts/release_control/work_claim.py --kind <KIND> --id <ID> --summary <SUMMARY> --agent-id <AGENT_ID> --pretty",
"python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty",
"python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty --lean",
"python3 scripts/release_control/worktree_base.py --base-branch <BASE_BRANCH> --pretty",
"python3 scripts/release_control/worktree_claim.py --kind <KIND> --id <ID> --summary <SUMMARY> --agent-id <AGENT_ID> --pretty",
"python3 scripts/release_control/worktree_finish.py --base-branch <BASE_BRANCH> --pretty",
@@ -123,6 +123,10 @@ class ControlPlaneAuditTest(unittest.TestCase):
"python3 scripts/release_control/work_claim.py --kind <KIND> --id <ID> --summary <SUMMARY> --agent-id <AGENT_ID> --pretty",
entrypoint["targeted_lookup_commands"],
)
self.assertIn(
"python3 scripts/release_control/subsystem_lookup.py <path> [<path> ...] --pretty --lean",
entrypoint["targeted_lookup_commands"],
)
self.assertIn("reserve exactly one", entrypoint["claim_rule"])
self.assertIn("record any remaining same-lane residual", entrypoint["claim_rule"])
self.assertIn(
+12 -2
View File
@@ -182,12 +182,22 @@ def parse_contract_text(rel: str, content: str) -> tuple[dict[str, Any], list[st
for heading in ("## Canonical Files", "## Shared Boundaries", "## Extension Points"):
if heading not in heading_positions:
continue
heading_line = heading_positions[heading] + 1
body_start = heading_positions[heading] + 1
body = section_body(lines, heading)
items = section_list_items(body)
for _, item in items:
for item_index, item in items:
line_number = body_start + item_index + 1
for token in re.findall(r"`([^`]+)`", item):
if looks_like_repo_path(token):
path_references.append({"heading": heading, "path": token})
path_references.append(
{
"heading": heading,
"path": token,
"line": line_number,
"heading_line": heading_line,
}
)
return {
"title": lines[0].strip() if lines else "",
@@ -76,9 +76,24 @@ Stable.
self.assertEqual(
parsed["path_references"],
[
{"heading": "## Canonical Files", "path": "internal/example/runtime.go"},
{"heading": "## Shared Boundaries", "path": "internal/shared/runtime.go"},
{"heading": "## Extension Points", "path": "internal/example/"},
{
"heading": "## Canonical Files",
"path": "internal/example/runtime.go",
"line": 17,
"heading_line": 15,
},
{
"heading": "## Shared Boundaries",
"path": "internal/shared/runtime.go",
"line": 21,
"heading_line": 19,
},
{
"heading": "## Extension Points",
"path": "internal/example/",
"line": 25,
"heading_line": 23,
},
],
)
@@ -138,8 +153,18 @@ Stable.
self.assertEqual(
matches[0]["matched_references"],
[
{"heading": "## Canonical Files", "path": "internal/example/runtime.go"},
{"heading": "## Extension Points", "path": "internal/example/"},
{
"heading": "## Canonical Files",
"path": "internal/example/runtime.go",
"line": 17,
"heading_line": 15,
},
{
"heading": "## Extension Points",
"path": "internal/example/",
"line": 25,
"heading_line": 23,
},
],
)
@@ -149,7 +174,12 @@ Stable.
self.assertEqual(
shared_matches[0]["matched_references"],
[
{"heading": "## Shared Boundaries", "path": "internal/shared/runtime.go"},
{
"heading": "## Shared Boundaries",
"path": "internal/shared/runtime.go",
"line": 21,
"heading_line": 19,
},
],
)
+178 -8
View File
@@ -20,6 +20,7 @@ from canonical_completion_guard import (
required_contract_updates,
subsystem_matches_path,
)
from subsystem_contracts import load_contract_index, referenced_contracts_for_path
from status_audit import audit_status_payload, load_status_payload
from registry_audit import load_registry_payload
@@ -82,6 +83,40 @@ def verification_requirements_for_test_path(rule: dict[str, Any], path: str) ->
return requirements
def ownership_basis_for_path(rule: dict[str, Any], path: str) -> dict[str, str] | None:
for exact_file in rule.get("owned_files", []):
if path == exact_file:
return {"type": "owned-file", "value": str(exact_file)}
for prefix in rule.get("owned_prefixes", []):
normalized = str(prefix).rstrip("/")
if path == normalized or path.startswith(str(prefix)):
return {"type": "owned-prefix", "value": str(prefix)}
return None
def normalized_reference_details(references: list[dict[str, Any]]) -> list[dict[str, Any]]:
normalized: list[dict[str, Any]] = []
for reference in references:
detail = {
"heading": str(reference.get("heading", "")),
"path": str(reference.get("path", "")),
}
if isinstance(reference.get("line"), int):
detail["line"] = int(reference["line"])
if isinstance(reference.get("heading_line"), int):
detail["heading_line"] = int(reference["heading_line"])
if detail not in normalized:
normalized.append(detail)
return sorted(
normalized,
key=lambda item: (
int(item.get("line", 0) or 0),
str(item.get("heading", "")).casefold(),
str(item.get("path", "")).casefold(),
),
)
def lane_context_for_rule(rule: dict[str, Any], status_report: dict[str, Any]) -> dict[str, Any] | None:
lane_id = str(rule.get("lane", "")).strip()
subsystem_id = str(rule.get("id", "")).strip()
@@ -115,10 +150,98 @@ def lane_context_for_rule(rule: dict[str, Any], status_report: dict[str, Any]) -
}
def lookup_paths(paths: list[str]) -> dict[str, Any]:
def compact_control_plane(control_plane: dict[str, Any]) -> dict[str, Any]:
active_target = dict(control_plane.get("active_target", {}))
return {
"active_profile_id": control_plane.get("active_profile_id"),
"active_target": {
key: active_target[key]
for key in ("id", "kind", "status", "completion_rule", "completion_met")
if key in active_target
},
}
def compact_lane_context(lane_context: dict[str, Any] | None) -> dict[str, Any] | None:
if lane_context is None:
return None
lane = lane_context.get("lane")
compact_lane = None
if isinstance(lane, dict):
compact_lane = {
key: lane.get(key)
for key in ("id", "name", "status", "derived_status", "completion_state", "gap", "repo_ids")
if key in lane
}
return {
"lane_id": lane_context.get("lane_id"),
"lane": compact_lane,
"open_decision_ids": [
str(decision.get("id"))
for decision in lane_context.get("open_decisions", [])
if isinstance(decision, dict) and decision.get("id")
],
"resolved_decision_ids": [
str(decision.get("id"))
for decision in lane_context.get("resolved_decisions", [])
if isinstance(decision, dict) and decision.get("id")
],
"release_gate_ids": [
str(gate.get("id"))
for gate in lane_context.get("release_gates", [])
if isinstance(gate, dict) and gate.get("id")
],
}
def compact_match(match: dict[str, Any]) -> dict[str, Any]:
compacted = dict(match)
compacted["lane_context"] = compact_lane_context(match.get("lane_context"))
return compacted
def compact_contract_update(contract_update: dict[str, Any]) -> dict[str, Any]:
compacted = dict(contract_update)
if compacted.get("matched_reference_details"):
compacted.pop("matched_references", None)
return compacted
def compact_lookup_result(result: dict[str, Any]) -> dict[str, Any]:
return {
"control_plane": compact_control_plane(result.get("control_plane", {})),
"status_audit_errors": list(result.get("status_audit_errors", [])),
"files": [
{
**entry,
"matches": [compact_match(match) for match in entry.get("matches", [])],
"dependent_contract_updates": [
compact_contract_update(contract)
for contract in entry.get("dependent_contract_updates", [])
],
}
for entry in result.get("files", [])
],
"impacted_subsystems": [
{
**entry,
"lane_context": compact_lane_context(entry.get("lane_context")),
}
for entry in result.get("impacted_subsystems", [])
],
"required_contract_updates": [
compact_contract_update(contract)
for contract in result.get("required_contract_updates", [])
],
"unowned_runtime_files": list(result.get("unowned_runtime_files", [])),
}
def lookup_paths(paths: list[str], *, lean: bool = False) -> dict[str, Any]:
normalized = [normalize_input_path(path) for path in paths if path.strip()]
rules = load_subsystem_rules()
rules_by_id = {str(rule["id"]): rule for rule in rules}
contract_index = load_contract_index()
registry_payload = load_registry_payload()
shared_ownership_by_path = {
str(entry["path"]): entry
@@ -137,6 +260,12 @@ def lookup_paths(paths: list[str]) -> dict[str, Any]:
classification = "ignored"
elif is_test_or_fixture(path):
classification = "test-or-fixture"
reference_matches_by_subsystem = {
str(contract["subsystem_id"]): normalized_reference_details(
list(contract.get("matched_references", []))
)
for contract in referenced_contracts_for_path(path, contract_index)
}
matches = []
if classification == "runtime":
@@ -151,6 +280,11 @@ def lookup_paths(paths: list[str]) -> dict[str, Any]:
"lane_context": lane_context_for_rule(rule, status_report),
"contract_update_required": True,
"proof_update_required": True,
"ownership_basis": ownership_basis_for_path(rule, path),
"matched_contract_references": reference_matches_by_subsystem.get(
str(rule["id"]),
[],
),
"verification_requirement": requirement,
}
)
@@ -164,6 +298,7 @@ def lookup_paths(paths: list[str]) -> dict[str, Any]:
"lane_context": lane_context_for_rule(rule, status_report),
"contract_update_required": False,
"proof_update_required": False,
"ownership_basis": ownership_basis_for_path(rule, path),
"verification_requirement": requirement,
}
)
@@ -200,7 +335,7 @@ def lookup_paths(paths: list[str]) -> dict[str, Any]:
}
)
return {
result = {
"control_plane": status_report.get("control_plane", {}),
"scope": status_report.get("scope", {}),
"status_summary": status_report.get("summary", {}),
@@ -210,6 +345,7 @@ def lookup_paths(paths: list[str]) -> dict[str, Any]:
"required_contract_updates": list(contract_updates.values()),
"unowned_runtime_files": unowned,
}
return compact_lookup_result(result) if lean else result
def parse_args(argv: list[str]) -> argparse.Namespace:
@@ -225,9 +361,24 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
action="store_true",
help="Print a concise human-readable summary instead of JSON.",
)
parser.add_argument(
"--lean",
action="store_true",
help="Return compact agent-facing context with exact contract reference lines.",
)
return parser.parse_args(argv)
def format_reference_detail(reference: dict[str, Any]) -> str:
heading = str(reference.get("heading", "")).strip() or "-"
path = str(reference.get("path", "")).strip()
line = reference.get("line")
line_suffix = f" @L{line}" if isinstance(line, int) else ""
if path:
return f"{heading}{line_suffix}: {path}"
return f"{heading}{line_suffix}"
def render_pretty(result: dict[str, Any]) -> str:
lines: list[str] = []
control_plane = result.get("control_plane", {})
@@ -257,12 +408,27 @@ def render_pretty(result: dict[str, Any]) -> str:
)
if lane_context and lane_context.get("lane"):
lane = lane_context["lane"]
gap = lane.get("gap")
gap_text = f"{gap:.0f}" if isinstance(gap, (int, float)) else "-"
open_decisions = lane_context.get("open_decisions", lane_context.get("open_decision_ids", []))
release_gates = lane_context.get("release_gates", lane_context.get("release_gate_ids", []))
lines.append(
f" lane {lane_context['lane_id']} "
f"gap={lane['gap']:.0f} derived={lane['derived_status']} "
f"gap={gap_text} derived={lane.get('derived_status') or '-'} "
f"repos={','.join(lane.get('repo_ids', [])) or '-'} "
f"open_decisions={len(lane_context['open_decisions'])} "
f"release_gates={len(lane_context['release_gates'])}"
f"open_decisions={len(open_decisions)} "
f"release_gates={len(release_gates)}"
)
if match.get("matched_contract_references"):
focus = ", ".join(
format_reference_detail(reference)
for reference in match["matched_contract_references"]
)
lines.append(f" contract focus: {focus}")
elif match.get("ownership_basis"):
basis = match["ownership_basis"]
lines.append(
f" ownership basis: {basis.get('type') or '-'} {basis.get('value') or '-'}"
)
if entry.get("shared_ownership"):
shared = entry["shared_ownership"]
@@ -273,8 +439,12 @@ def render_pretty(result: dict[str, Any]) -> str:
lines.append(
f" - also update {contract['subsystem']} contract -> {contract['contract']}"
)
for reference in contract.get("matched_references", []):
lines.append(f" referenced by {reference}")
if contract.get("matched_reference_details"):
for reference in contract["matched_reference_details"]:
lines.append(f" referenced by {format_reference_detail(reference)}")
else:
for reference in contract.get("matched_references", []):
lines.append(f" referenced by {reference}")
if entry["classification"] == "runtime" and not entry["matches"]:
lines.append(" - no owning subsystem rule matched")
for path in result["unowned_runtime_files"]:
@@ -295,7 +465,7 @@ def main(argv: list[str] | None = None) -> int:
print("no file paths provided", file=sys.stderr)
return 2
result = lookup_paths(paths)
result = lookup_paths(paths, lean=args.lean)
output = render_pretty(result) if args.pretty else json.dumps(result, indent=2, sort_keys=True)
print(output)
return 0
@@ -2,7 +2,7 @@ import unittest
from pathlib import Path
from canonical_completion_guard import REPO_ROOT
from subsystem_lookup import lookup_paths
from subsystem_lookup import lookup_paths, parse_args, render_pretty
RECOVERY_PRODUCT_SURFACE_EXACT_FILES = [
@@ -39,6 +39,12 @@ PLATFORM_CONNECTIONS_WORKSPACE_EXACT_FILES = [
class SubsystemLookupTest(unittest.TestCase):
def test_parse_args_accepts_lean_flag(self) -> None:
args = parse_args(["internal/api/ai_handler.go", "--pretty", "--lean"])
self.assertEqual(args.paths, ["internal/api/ai_handler.go"])
self.assertTrue(args.pretty)
self.assertTrue(args.lean)
def test_lookup_paths_reports_multiple_subsystems_for_shared_runtime_file(self) -> None:
result = lookup_paths(["internal/api/resources.go"])
impacted = {entry["subsystem"] for entry in result["impacted_subsystems"]}
@@ -4441,11 +4447,106 @@ class SubsystemLookupTest(unittest.TestCase):
{contract["subsystem"] for contract in file_entry["dependent_contract_updates"]},
{"monitoring"},
)
monitoring_contract = file_entry["dependent_contract_updates"][0]
self.assertEqual(
monitoring_contract["matched_reference_details"],
[
{
"heading": "## Canonical Files",
"path": "internal/unifiedresources/views.go",
"line": 32,
"heading_line": 23,
},
{
"heading": "## Extension Points",
"path": "internal/unifiedresources/views.go",
"line": 46,
"heading_line": 42,
},
],
)
self.assertEqual(
{contract["subsystem"] for contract in result["required_contract_updates"]},
{"monitoring", "unified-resources"},
)
def test_lookup_paths_reports_contract_focus_for_shared_api_runtime_file(self) -> None:
result = lookup_paths(["internal/api/ai_handler.go"])
file_entry = result["files"][0]
by_subsystem = {match["subsystem"]: match for match in file_entry["matches"]}
self.assertEqual(
by_subsystem["ai-runtime"]["matched_contract_references"],
[
{
"heading": "## Canonical Files",
"path": "internal/api/ai_handler.go",
"line": 24,
"heading_line": 21,
},
{
"heading": "## Shared Boundaries",
"path": "internal/api/ai_handler.go",
"line": 45,
"heading_line": 41,
},
{
"heading": "## Extension Points",
"path": "internal/api/ai_handler.go",
"line": 52,
"heading_line": 49,
},
],
)
self.assertEqual(
by_subsystem["api-contracts"]["matched_contract_references"],
[
{
"heading": "## Shared Boundaries",
"path": "internal/api/ai_handler.go",
"line": 108,
"heading_line": 77,
},
{
"heading": "## Extension Points",
"path": "internal/api/ai_handler.go",
"line": 167,
"heading_line": 125,
},
{
"heading": "## Extension Points",
"path": "internal/api/ai_handler.go",
"line": 168,
"heading_line": 125,
},
],
)
self.assertEqual(
by_subsystem["api-contracts"]["ownership_basis"],
{"type": "owned-prefix", "value": "internal/api/"},
)
def test_lookup_paths_lean_mode_trims_status_payload_but_keeps_contract_focus(self) -> None:
result = lookup_paths(["internal/api/ai_handler.go"], lean=True)
self.assertNotIn("scope", result)
self.assertNotIn("status_summary", result)
file_entry = result["files"][0]
by_subsystem = {match["subsystem"]: match for match in file_entry["matches"]}
api_match = by_subsystem["api-contracts"]
self.assertEqual(api_match["lane_context"]["lane_id"], "L6")
self.assertEqual(api_match["lane_context"]["open_decision_ids"], [])
self.assertIn("paid-feature-entitlement-gating", api_match["lane_context"]["release_gate_ids"])
self.assertEqual(
[reference["line"] for reference in api_match["matched_contract_references"]],
[108, 167, 168],
)
def test_render_pretty_shows_contract_focus_for_lean_lookup(self) -> None:
rendered = render_pretty(lookup_paths(["internal/api/ai_handler.go"], lean=True))
self.assertIn("contract focus: ## Shared Boundaries @L108: internal/api/ai_handler.go", rendered)
self.assertIn("contract focus: ## Canonical Files @L24: internal/api/ai_handler.go", rendered)
def test_lookup_paths_maps_unified_agent_runtime_to_agent_lifecycle(self) -> None:
result = lookup_paths(["internal/hostagent/agent.go"])
self.assertEqual(result["unowned_runtime_files"], [])