Require demand_evidence on candidate_lane registration

Every candidate_lanes record now carries demand_evidence: dated pointers
to the demand signals justifying the proposed lane (issues, discussions,
support threads, telemetry findings, or a demand-ledger entry in
pulse-pro/FEATURE_REQUESTS.md), or an explicit named-bet declaration for
maintainer-originated lanes with no external signal yet. The schema
requires the field, status_audit rejects missing or empty lists and
surfaces the evidence in the candidate_lanes and candidate_lane_queue
pretty blocks, and the canonical development protocol documents the
rule. No data migration needed; candidate_lanes is currently empty.
This commit is contained in:
rcourtman
2026-08-24 15:59:32 +01:00
parent 7e363e40e7
commit 9c7cb9ed9b
4 changed files with 82 additions and 1 deletions
@@ -355,6 +355,13 @@ Every substantial task must finish by checking these questions:
When the intended destination is already clear, also add or update the
matching `candidate_lanes` record so the lane-expansion plan is typed,
machine-readable, and pointed at the owning control-plane target.
Every `candidate_lanes` record must carry `demand_evidence`: dated
pointers to the demand signals that justify the lane (issue, discussion,
support-thread, or telemetry references, or a demand-ledger entry in
`pulse-pro/FEATURE_REQUESTS.md`), or an explicit `named-bet: <rationale>`
entry when the lane is maintainer-originated with no external signal
yet. `status_audit.py` rejects records without it; a named bet is
legitimate, an unlabelled hunch is not.
Once a lane-shaping gap (`new-lane`, `lane-split`, or `lane-expansion`)
has a typed `candidate_lanes` record, move that `coverage_gaps` entry to
`planned`; do not mark such a gap `planned` before the matching
+11 -1
View File
@@ -1247,7 +1247,8 @@
"target_id",
"current_lane_ids",
"coverage_gap_ids",
"subsystem_ids"
"subsystem_ids",
"demand_evidence"
],
"properties": {
"id": {
@@ -1303,6 +1304,15 @@
"type": "string",
"minLength": 1
}
},
"demand_evidence": {
"type": "array",
"minItems": 1,
"uniqueItems": true,
"items": {
"type": "string",
"minLength": 1
}
}
}
},
+11
View File
@@ -2086,6 +2086,13 @@ def validate_candidate_lanes(
current_lane_refs = _require_string_list(raw, "current_lane_ids", errors, context=context)
gap_refs = _require_string_list(raw, "coverage_gap_ids", errors, context=context)
subsystem_refs = _require_string_list(raw, "subsystem_ids", errors, context=context)
demand_evidence = _require_string_list(raw, "demand_evidence", errors, context=context)
if raw.get("demand_evidence") == []:
errors.append(
f"{context}.demand_evidence must cite at least one demand signal "
"(issue/discussion/support/telemetry pointer, or demand-ledger entry) "
"or carry a 'named-bet: <rationale>' declaration"
)
if recorded_at:
_validate_date(recorded_at, errors, context=f"{context}.recorded_at")
@@ -2194,6 +2201,7 @@ def validate_candidate_lanes(
"current_lane_ids": current_lane_refs,
"coverage_gap_ids": gap_refs,
"subsystem_ids": subsystem_refs,
"demand_evidence": demand_evidence,
"repo_ids": sorted(repo_ids, key=_repo_sort_key),
"cross_repo": len(repo_ids) > 1,
}
@@ -3217,6 +3225,7 @@ def audit_status_payload(
"current_lane_ids": list(candidate["current_lane_ids"]),
"repo_ids": list(candidate["repo_ids"]),
"subsystem_ids": list(candidate["subsystem_ids"]),
"demand_evidence": list(candidate["demand_evidence"]),
"cross_repo": bool(candidate["cross_repo"]),
"claim_ids": sorted(
[
@@ -3572,6 +3581,7 @@ def render_pretty(report: dict[str, Any]) -> str:
f"subsystems={','.join(candidate['subsystem_ids']) or '-'}"
)
lines.append(f" name={candidate['name']}")
lines.append(f" demand_evidence={'; '.join(candidate['demand_evidence'])}")
lines.append(f" {candidate['summary']}")
if report.get("work_claims"):
lines.append("work_claims:")
@@ -3597,6 +3607,7 @@ def render_pretty(report: dict[str, Any]) -> str:
f"coverage_gaps={','.join(item['coverage_gap_ids']) or '-'}"
)
lines.append(f" name={item['name']}")
lines.append(f" demand_evidence={'; '.join(item['demand_evidence'])}")
if item["claim_ids"]:
lines.append(
f" claims={','.join(item['claim_ids'])} agents={','.join(item['claim_agent_ids'])}"
@@ -115,6 +115,7 @@ def candidate_lane(
subsystem_ids: list[str] | None = None,
status: str = "planned",
target_id: str = "v6-product-lane-expansion",
demand_evidence: list[str] | None = None,
) -> dict[str, object]:
return {
"id": candidate_id,
@@ -126,6 +127,10 @@ def candidate_lane(
"current_lane_ids": list(current_lane_ids or ["L1"]),
"coverage_gap_ids": list(gap_ids or ["core-monitoring-product-lane"]),
"subsystem_ids": list(subsystem_ids or []),
"demand_evidence": list(
demand_evidence
or ["named-bet: promote the discovered surface into a governed lane"]
),
}
@@ -1257,6 +1262,10 @@ class StatusAuditTest(unittest.TestCase):
pretty,
)
self.assertIn("name=Core monitoring runtime", pretty)
self.assertIn(
"demand_evidence=named-bet: promote the discovered surface into a governed lane",
pretty,
)
self.assertIn("candidate_lane_queue:", pretty)
self.assertIn(
"rank=1 candidate=core-monitoring-runtime impact=5 target=v6-product-lane-expansion",
@@ -1608,6 +1617,50 @@ class StatusAuditTest(unittest.TestCase):
report["errors"],
)
def test_candidate_lane_requires_demand_evidence(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
pulse = Path(tmp) / "pulse"
pulse.mkdir()
write_file(pulse, "docs/lane-proof.md")
write_file(pulse, "docs/proof_test.go")
write_file(pulse, "docs/hybrid_test.go")
write_file(pulse, "docs/coverage-gap.md")
missing = candidate_lane()
del missing["demand_evidence"]
empty = candidate_lane(
candidate_id="core-monitoring-runtime-empty",
name="Core monitoring runtime empty",
gap_ids=["core-monitoring-product-lane-empty"],
)
empty["demand_evidence"] = []
payload = base_payload(
coverage_gaps=[
coverage_gap(status="planned"),
coverage_gap(gap_id="core-monitoring-product-lane-empty", status="planned"),
],
candidate_lanes=[missing, empty],
)
with mock.patch.dict(os.environ, {"PULSE_REPO_ROOT_PULSE": str(pulse)}, clear=False), mock.patch(
"status_audit.load_subsystem_rules",
return_value=[],
):
report = audit_status_payload(payload)
self.assertIn(
"candidate_lanes[0] missing list demand_evidence",
report["errors"],
)
self.assertTrue(
any(
error.startswith("candidate_lanes[1].demand_evidence must cite at least one demand signal")
for error in report["errors"]
),
report["errors"],
)
def test_candidate_lane_rejects_unknown_or_completed_target(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
pulse = Path(tmp) / "pulse"