REUSE-WP-0018 T01/T02/T04/T06: plan-check deterministic matching + State Hub bridge
Some checks failed
ci / validate-registry (push) Has been cancelled

T01: specs/PlanCheck.md design doc, plan-check-result.schema.json and
reuse-event.schema.json (the latter shared with WP-0019's reuse telemetry).

T02: reuse_surface/plan_check.py + 'reuse-surface plan-check' CLI command.
Deterministic token-Jaccard matching against registry/indexes/federated.yaml
(reuses overlaps.py's TOKEN_RE rather than a second scoring method), with
reuse/extend/new verdicts, markdown and --format json output, staleness
warning, and --record-outcome JSONL telemetry.

T04: reuse_surface/statehub_bridge.py bridges plan-check 'new' verdicts to
State Hub capability requests (--file-request) and surfaces open requests
with no matching capability (report gaps --check-capability-requests, opt-in
to stay offline-safe). Verified against the live local State Hub API;
status field (not catalog_entry_id presence) is the correct open/closed
signal, and the list endpoint needs a longer timeout than the health check
(~7s observed with 5 rows).

T06: docs (tools/README.md, RegistryFederation.md, SCOPE.md,
IntentScopeGapAnalysis.md priority 29) and an informational CI smoke step.

T03 (LLM rerank) not started -- llm-connect isn't running on this
workstation. T05 (ecosystem rollout) remains blocked: WP-0017 has drafted
entries for all 61 repos but they're still local-only pending its own T05
push/publish pass, so the federated index isn't yet worth rolling out
plan-check as ecosystem convention.

16 new tests, all mocked -- no network calls in the default test run.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-07 00:57:18 +02:00
parent 2a6818d4fe
commit e72966a658
15 changed files with 1179 additions and 29 deletions

198
tests/test_plan_check.py Normal file
View File

@@ -0,0 +1,198 @@
from __future__ import annotations
import json
from reuse_surface.plan_check import (
MatchQuery,
load_query_from_intent,
load_query_from_workplan,
maybe_file_capability_request,
match_query,
record_outcome,
run_plan_check,
verdict_for_score,
)
SAMPLE_CAPABILITIES = [
{
"id": "capability.infotech.issue-tracking",
"name": "Universal Issue Tracking Coordination",
"summary": "Unified interface for issue tracking coordination across Gitea, GitHub, GitLab.",
"vector": "D4 / A2 / C2 / R1",
"owner": "issue-core",
"tags": ["issue-tracking", "coordination"],
},
{
"id": "capability.audit.event-retain",
"name": "Audit Event Retention",
"summary": "Collect, normalize, retain, and search audit events with integrity evidence across tenants.",
"vector": "D4 / A2 / C2 / R1",
"owner": "audit-core",
"tags": ["audit"],
},
]
def test_verdict_thresholds():
assert verdict_for_score(0.5) == "reuse"
assert verdict_for_score(0.3) == "extend"
assert verdict_for_score(0.1) == "new"
assert verdict_for_score(0.45) == "reuse"
assert verdict_for_score(0.22) == "extend"
def test_match_query_finds_close_match():
query = load_query_from_intent(
"unified issue tracking coordination across Gitea GitHub GitLab"
)
matches = match_query(query, SAMPLE_CAPABILITIES)
assert matches
assert matches[0].id == "capability.infotech.issue-tracking"
assert matches[0].score > 0.3
def test_match_query_empty_when_no_overlap():
query = load_query_from_intent("something entirely unrelated xyzzy plugh")
matches = match_query(query, SAMPLE_CAPABILITIES)
assert matches == []
def test_match_query_no_capabilities():
query = load_query_from_intent("anything")
assert match_query(query, []) == []
def test_load_query_from_workplan(tmp_path):
workplan = tmp_path / "TEST-WP-0001-thing.md"
workplan.write_text(
"""---
id: TEST-WP-0001
title: "Unified issue tracking coordination"
status: proposed
---
# Unified issue tracking coordination
## Core Idea
Build a single interface for issue tracking across Gitea, GitHub, and GitLab.
"""
)
query = load_query_from_workplan(workplan)
assert query.source == "workplan"
assert query.workplan_id == "TEST-WP-0001"
assert "issue tracking" in query.text.lower()
matches = match_query(query, SAMPLE_CAPABILITIES)
assert matches[0].id == "capability.infotech.issue-tracking"
def test_run_plan_check_reuse_verdict(monkeypatch):
monkeypatch.setattr(
"reuse_surface.plan_check.load_federated_capabilities",
lambda: (SAMPLE_CAPABILITIES, "2026-07-06"),
)
query = MatchQuery(
source="intent",
text="unified issue tracking coordination gitea github gitlab",
tokens={"unified", "issue", "tracking", "coordination", "gitea", "github", "gitlab"},
)
result = run_plan_check(query)
assert result["verdict"] == "reuse"
assert result["matches"][0]["id"] == "capability.infotech.issue-tracking"
assert result["federated_index_stale_warning"] is None
def test_run_plan_check_stale_warning(monkeypatch):
monkeypatch.setattr(
"reuse_surface.plan_check.load_federated_capabilities",
lambda: (SAMPLE_CAPABILITIES, "2020-01-01"),
)
query = load_query_from_intent("nothing matches this at all zzz")
result = run_plan_check(query)
assert result["verdict"] == "new"
assert "days old" in result["federated_index_stale_warning"]
def test_record_outcome_appends_jsonl(tmp_path, monkeypatch):
telemetry_path = tmp_path / "plan-check-events.jsonl"
monkeypatch.setattr("reuse_surface.plan_check.TELEMETRY_PATH", telemetry_path)
result = {
"verdict": "reuse",
"matches": [{"id": "capability.infotech.issue-tracking"}],
}
record_outcome(result, "reused", consumer_repo="some-repo")
lines = telemetry_path.read_text().splitlines()
assert len(lines) == 1
event = json.loads(lines[0])
assert event["consumer_repo"] == "some-repo"
assert event["capability_id"] == "capability.infotech.issue-tracking"
assert event["outcome"] == "reused"
assert event["source"] == "plan-check"
def test_maybe_file_capability_request_only_on_new_verdict(monkeypatch):
called = []
monkeypatch.setattr(
"reuse_surface.plan_check.file_capability_request",
lambda **kwargs: called.append(kwargs) or {"id": "req-1"},
)
reuse_result = {"verdict": "reuse", "query": {"text": "x", "workplan_id": None}}
assert maybe_file_capability_request(
reuse_result, requesting_domain="infotech", requesting_agent="a"
) is None
assert called == []
new_result = {"verdict": "new", "query": {"text": "some new intent", "workplan_id": None}}
filed = maybe_file_capability_request(
new_result, requesting_domain="infotech", requesting_agent="a"
)
assert filed == {"id": "req-1"}
assert called[0]["requesting_domain"] == "infotech"
def test_cmd_plan_check_file_request_flag(monkeypatch):
from reuse_surface.cli import main
monkeypatch.setattr(
"reuse_surface.plan_check.load_federated_capabilities",
lambda: (SAMPLE_CAPABILITIES, "2026-07-06"),
)
filed_calls = []
monkeypatch.setattr(
"reuse_surface.cli.maybe_file_capability_request",
lambda result, **kwargs: filed_calls.append(kwargs) or {"id": "req-9"},
)
exit_code = main(
["plan-check", "--intent", "totally unrelated xyzzy plugh", "--file-request"]
)
assert exit_code == 0
assert filed_calls
def test_cmd_plan_check_intent_json(monkeypatch):
from reuse_surface.cli import main
monkeypatch.setattr(
"reuse_surface.plan_check.load_federated_capabilities",
lambda: (SAMPLE_CAPABILITIES, "2026-07-06"),
)
exit_code = main(
["plan-check", "--intent", "issue tracking gitea github gitlab", "--format", "json"]
)
assert exit_code == 0
def test_cmd_plan_check_requires_input():
from reuse_surface.cli import main
exit_code = main(["plan-check"])
assert exit_code == 1
def test_cmd_plan_check_rejects_both_inputs(tmp_path):
from reuse_surface.cli import main
workplan = tmp_path / "x.md"
workplan.write_text("---\nid: X\n---\nbody")
exit_code = main(["plan-check", str(workplan), "--intent", "also this"])
assert exit_code == 1

View File

@@ -158,6 +158,29 @@ def test_cmd_report_gaps_json(monkeypatch):
assert exit_code == 0
def test_cmd_report_gaps_check_capability_requests_unreachable(monkeypatch):
from reuse_surface.cli import main
monkeypatch.setattr(
"reuse_surface.cli.list_open_capability_requests", lambda: None
)
exit_code = main(["report", "gaps", "--check-capability-requests"])
assert exit_code == 0
def test_cmd_report_gaps_check_capability_requests_json(monkeypatch):
from reuse_surface.cli import main
monkeypatch.setattr(
"reuse_surface.cli.list_open_capability_requests",
lambda: [{"id": "r1", "title": "Need X", "requesting_domain_slug": "infotech"}],
)
exit_code = main(
["report", "gaps", "--check-capability-requests", "--format", "json"]
)
assert exit_code == 0
def test_cmd_report_cohorts_markdown(monkeypatch):
from reuse_surface.cli import main

View File

@@ -0,0 +1,66 @@
from __future__ import annotations
import urllib.error
from reuse_surface import statehub_bridge
def test_state_hub_reachable_true(monkeypatch):
monkeypatch.setattr(statehub_bridge, "_request", lambda *a, **k: {"status": "ok"})
assert statehub_bridge.state_hub_reachable() is True
def test_state_hub_reachable_false_on_error(monkeypatch):
def raise_error(*args, **kwargs):
raise urllib.error.URLError("no route")
monkeypatch.setattr(statehub_bridge, "_request", raise_error)
assert statehub_bridge.state_hub_reachable() is False
def test_file_capability_request_skips_when_unreachable(monkeypatch):
monkeypatch.setattr(statehub_bridge, "state_hub_reachable", lambda base_url=None: False)
result = statehub_bridge.file_capability_request(
title="t", description="d", requesting_domain="infotech", requesting_agent="a"
)
assert result is None
def test_file_capability_request_posts_body(monkeypatch):
captured = {}
def fake_request(method, path, base_url, *, body=None, params=None, timeout=None):
captured["method"] = method
captured["path"] = path
captured["body"] = body
return {"id": "abc-123", **body}
monkeypatch.setattr(statehub_bridge, "state_hub_reachable", lambda base_url=None: True)
monkeypatch.setattr(statehub_bridge, "_request", fake_request)
result = statehub_bridge.file_capability_request(
title="Need X", description="desc", requesting_domain="infotech", requesting_agent="reuse-surface"
)
assert captured["method"] == "POST"
assert captured["path"] == "/capability-requests/"
assert captured["body"]["title"] == "Need X"
assert result["id"] == "abc-123"
def test_list_open_capability_requests_filters_terminal_status(monkeypatch):
monkeypatch.setattr(statehub_bridge, "state_hub_reachable", lambda base_url=None: True)
monkeypatch.setattr(
statehub_bridge,
"_request",
lambda *a, **k: [
{"id": "1", "status": "requested"},
{"id": "2", "status": "completed"},
{"id": "3", "status": "requested", "catalog_entry_id": "routed-but-open"},
],
)
result = statehub_bridge.list_open_capability_requests()
assert {r["id"] for r in result} == {"1", "3"}
def test_list_open_capability_requests_none_when_unreachable(monkeypatch):
monkeypatch.setattr(statehub_bridge, "state_hub_reachable", lambda base_url=None: False)
assert statehub_bridge.list_open_capability_requests() is None