diff --git a/SCOPE.md b/SCOPE.md index 53489b7..6fb47fb 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -55,7 +55,7 @@ The two evaluation modes: scheduled datetime; event-type subscription via NATS; manual one-shot API trigger; one-shot schedule smoke tests for recurring definitions. - **Context resolution adapters**: repo-scoping (repository capability queries), - State Hub (domain/workstream state, SBOM status, daily triage digest, coding + State Hub (domain/workplan state, SBOM status, daily triage digest, coding retro read model), and ops inventory (bounded HTTP/HTTPS probes of a non-secret service inventory). The adapter registry is extensible for other sources. @@ -218,7 +218,7 @@ workplan explicitly rehomes execution responsibility. One boundary nuance is now explicit: activity-core may post State Hub progress events as a configured report or evidence sink. That is acceptable because it records the result of an activity-core activation; it is not ownership of State -Hub state, task lifecycle, or workstream planning. +Hub state, task lifecycle, or workplan planning. The main drift risk is convenience creep: adding direct task tracking, project-phase state, or bespoke operational scripts because the Temporal @@ -277,7 +277,7 @@ new one-off control paths. - **Coordinates with**: the state hub delegates maintenance automations to activity-core by publishing lifecycle events or by being resolved as context. activity-core may post progress events as report/evidence outputs, but it - does not own State Hub task/workstream state. + does not own State Hub task/workplan state. --- diff --git a/k8s/railiance/21-custodian-runtime.yaml b/k8s/railiance/21-custodian-runtime.yaml index a73dec6..e9638fd 100644 --- a/k8s/railiance/21-custodian-runtime.yaml +++ b/k8s/railiance/21-custodian-runtime.yaml @@ -61,13 +61,13 @@ data: 3. `GET http://127.0.0.1:8000/messages/?to_agent=hub&unread_only=true` 4. `GET http://127.0.0.1:8000/state/next_steps` 5. `GET http://127.0.0.1:8000/workstreams/workplan-index` - 6. Local workplan files for candidate workstreams that need closer inspection + 6. Local workplan files for candidate workplans that need closer inspection 7. Read-only `git status --short` for repos named in top recommendations Optional enrichment: - `GET /tasks/?workplan_id=` for a top-ranked workplan (legacy alias: workstream_id) - - `GET /progress/?workstream_id=&limit=5` for staleness confidence + - `GET /progress/?workplan_id=&limit=5` for staleness confidence (legacy alias: workstream_id) - State Hub domain summaries through MCP when available ## Candidate Set @@ -93,9 +93,9 @@ data: | Active with no recent progress | no progress in 14 days, or 7 days for high-priority items | `revisit` | | Large unstarted plan | `tasks_total >= 6`, `tasks_done == 0`, `tasks_in_progress == 0` | `split` or `revisit` | | Near complete | `tasks_done / tasks_total >= 0.75` and `tasks_todo <= 3` | `close-out` or `work-next` | - | Zero parsed tasks | active or blocked workstream with `tasks_total == 0` | `needs-consistency-sync` | - | Blocked but dependency appears closed | blocked workstream with cleared dependency or empty blocked reason | `revisit` | - | Open decision blocks execution | blocking decision connected to workstream/task | `needs-human` | + | Zero parsed tasks | active or blocked workplan with `tasks_total == 0` | `needs-consistency-sync` | + | Blocked but dependency appears closed | blocked workplan with cleared dependency or empty blocked reason | `revisit` | + | Open decision blocks execution | blocking decision connected to workplan/task | `needs-human` | | File-backed mismatch | missing index entry, stale `updated`, `needs_review`, or status disagreement | `needs-consistency-sync` | | Too broad for one session | many unrelated task clusters in one workplan | `split` | | Safety-sensitive | security, identity, secrets, legal, external reputation, money | `needs-human` | @@ -120,7 +120,7 @@ data: - dependency-chain foundations: Railiance, identity, secrets, backup, HA - explicit `planning_priority: high` or low `planning_order` - active repo/domain goals - - closing a workstream that is already consuming coordination attention + - closing a workplan that is already consuming coordination attention ### Time Criticality @@ -144,7 +144,7 @@ data: Score higher when the work unlocks: - - multiple downstream workstreams + - multiple downstream workplans - agent autonomy or fewer repeated setup steps - reliable cross-repo operation - future daily triage quality @@ -201,7 +201,7 @@ data: ## Snapshot - Generated at: - - Workstreams: active, blocked, finished + - Workplans: active, blocked, finished - Tasks: todo, in progress, blocked - Decisions: open, escalated - Inbox: unread hub messages @@ -211,7 +211,7 @@ data: | Rank | Action | Candidate | WSJF | Confidence | Why now | |------|--------|-----------|------|------------|---------| - | 1 | work-next | | 0.0 | high | | + | 1 | work-next | | 0.0 | high | | ## Loose Ends @@ -240,7 +240,7 @@ data: When the State Hub API is reachable, append one progress event for the review: - `topic_id`: Custodian topic id when known - - `workstream_id`: `99993845-be6a-401d-be98-f8107014abed` + - `workplan_id`: `99993845-be6a-401d-be98-f8107014abed` (legacy alias: workstream_id) - `event_type`: `daily_triage` - `summary`: one sentence summarizing the top recommendation and loose-end count - `detail`: include top three recommendations, input health, and any human gates diff --git a/src/activity_core/context_resolvers/state_hub.py b/src/activity_core/context_resolvers/state_hub.py index b899758..3cbcb4f 100644 --- a/src/activity_core/context_resolvers/state_hub.py +++ b/src/activity_core/context_resolvers/state_hub.py @@ -10,7 +10,7 @@ Supported queries: - workplan_index: GET {STATE_HUB_URL}/workstreams/workplan-index - hub_inbox: GET {STATE_HUB_URL}/messages/?to_agent=hub&unread_only=true - pending_decisions: GET {STATE_HUB_URL}/decisions/?status=open (topic_id/ - workstream_id/decision_type passed through as given) + workplan_id or legacy workstream_id/decision_type passed through) - coding_retro: latest /progress/ item with event_type=coding_retro - daily_triage_digest: curated scalar JSON digest for daily WSJF triage - recently_on_scope_hourly: POST {STATE_HUB_URL}/recently-on-scope/hourly @@ -199,7 +199,9 @@ def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]: } baseline = params.get("baseline") or {} - expected_workstreams = int(baseline.get("workstreams", 640)) + expected_workplan_total = int( + baseline.get("workplans", baseline.get("workstreams", 640)) + ) expected_tasks = int(baseline.get("tasks", 4002)) expected_topics = int(baseline.get("topics", 14)) sweep_limit = _bounded_int(params.get("sweep_limit", 6), default=6, minimum=1, maximum=24) @@ -215,11 +217,12 @@ def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]: summary = _fetch_json("/state/summary") totals = (summary or {}).get("totals") or {} - ws_total = int((totals.get("workstreams") or {}).get("total", -1)) + workplan_totals = totals.get("workplans") or totals.get("workstreams") or {} + ws_total = int(workplan_totals.get("total", -1)) task_total = int((totals.get("tasks") or {}).get("total", -1)) topic_total = int((totals.get("topics") or {}).get("total", -1)) totals_pass = ( - ws_total == expected_workstreams + ws_total == expected_workplan_total and task_total == expected_tasks and topic_total == expected_topics ) @@ -259,11 +262,13 @@ def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]: "health": {"pass": health_pass, "status": health.get("status") if isinstance(health, dict) else None}, "totals": { "pass": totals_pass, + "workplans": ws_total, "workstreams": ws_total, "tasks": task_total, "topics": topic_total, "expected": { - "workstreams": expected_workstreams, + "workplans": expected_workplan_total, + "workstreams": expected_workplan_total, "tasks": expected_tasks, "topics": expected_topics, }, diff --git a/src/activity_core/ops_evidence_sinks.py b/src/activity_core/ops_evidence_sinks.py index 9e503d0..1eb145c 100644 --- a/src/activity_core/ops_evidence_sinks.py +++ b/src/activity_core/ops_evidence_sinks.py @@ -10,7 +10,7 @@ from typing import Any import httpx from activity_core.context_resolvers.ops_inventory import _sanitize_url -from activity_core.state_hub_write import idempotency_headers +from activity_core.state_hub_write import apply_progress_scope_fields, idempotency_headers _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000" _INTER_HUB_SINK_TYPES = { @@ -149,9 +149,7 @@ def _post_state_hub_progress( "probe": compact, }, } - for key in ("topic_id", "workstream_id", "task_id", "decision_id"): - if sink.get(key): - body[key] = sink[key] + apply_progress_scope_fields(body, sink) resp = httpx.post( f"{base_url}/progress/", @@ -529,7 +527,8 @@ def _phase5_summary_text(result: dict[str, Any]) -> str: mode = "closeout" if result.get("closeout") else "daily" return ( f"Phase 5 stabilization {mode}: {status}; " - f"totals {totals.get('workstreams', '?')}/{totals.get('tasks', '?')}/" + f"workplan totals {totals.get('workplans', totals.get('workstreams', '?'))}/" + f"{totals.get('tasks', '?')}/" f"{totals.get('topics', '?')}; sweeps sampled={sweeps.get('sampled', 0)}; " f"daily_triage age_h={triage.get('age_hours', '?')}" ) diff --git a/src/activity_core/report_sinks.py b/src/activity_core/report_sinks.py index 9b64c52..592ea46 100644 --- a/src/activity_core/report_sinks.py +++ b/src/activity_core/report_sinks.py @@ -16,7 +16,7 @@ from activity_core.runtime_paths import ( custodian_repo_root, resolve_runtime_path, ) -from activity_core.state_hub_write import idempotency_headers +from activity_core.state_hub_write import apply_progress_scope_fields, idempotency_headers _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000" @@ -146,9 +146,7 @@ def _post_state_hub_progress( body["detail"]["working_memory_status"] = report_entry.get( "working_memory_status" ) - for key in ("topic_id", "workstream_id", "task_id", "decision_id"): - if sink.get(key): - body[key] = sink[key] + apply_progress_scope_fields(body, sink) resp = httpx.post( f"{base_url}/progress/", diff --git a/src/activity_core/schedule_health.py b/src/activity_core/schedule_health.py index d363812..77b2244 100644 --- a/src/activity_core/schedule_health.py +++ b/src/activity_core/schedule_health.py @@ -139,6 +139,7 @@ def post_missed_fire_alert( state_hub_url: str | None = None, author: str = "activity-core", topic_id: str | None = None, + workplan_id: str | None = None, workstream_id: str | None = None, timeout_seconds: float = 10.0, ) -> dict[str, Any]: @@ -174,8 +175,10 @@ def post_missed_fire_alert( } if topic_id: body["topic_id"] = topic_id - if workstream_id: - body["workstream_id"] = workstream_id + scope_id = workplan_id or workstream_id + if scope_id: + body["workplan_id"] = scope_id + body["workstream_id"] = scope_id # Dedup repeated alerts for the same missed window (same schedule + last fire). last_fired = health.last_fired_at.isoformat() if health.last_fired_at else "none" diff --git a/src/activity_core/state_hub_write.py b/src/activity_core/state_hub_write.py index 2b45138..a4eae34 100644 --- a/src/activity_core/state_hub_write.py +++ b/src/activity_core/state_hub_write.py @@ -15,9 +15,26 @@ ignored by State Hub versions that do not yet honour it. from __future__ import annotations +from typing import Any + IDEMPOTENCY_HEADER = "Idempotency-Key" +def apply_progress_scope_fields(body: dict[str, Any], source: dict[str, Any]) -> None: + """Copy topic/workplan/task/decision scope from a sink config into a progress body. + + Prefers ``workplan_id``; dual-writes ``workstream_id`` for wire-compat until + legacy-meter retires the alias. + """ + workplan_id = source.get("workplan_id") or source.get("workstream_id") + if workplan_id: + body["workplan_id"] = workplan_id + body["workstream_id"] = workplan_id + for key in ("topic_id", "task_id", "decision_id"): + if source.get(key): + body[key] = source[key] + + def idempotency_key(*parts: str | None) -> str: """Build a stable, header-safe idempotency key from identity parts. diff --git a/tests/test_ops_evidence_sinks.py b/tests/test_ops_evidence_sinks.py index f1b96c2..24dfa05 100644 --- a/tests/test_ops_evidence_sinks.py +++ b/tests/test_ops_evidence_sinks.py @@ -123,6 +123,7 @@ def test_state_hub_progress_sink_posts_compact_probe_summary(monkeypatch) -> Non ] body = posts[0]["json"] assert body["summary"] == "Ops inventory probe: 1 ok, 0 degraded, 0 down, 1 skipped" + assert body["workplan_id"] == "workstream-1" assert body["workstream_id"] == "workstream-1" assert body["task_id"] == "task-1" assert body["detail"]["activity_core_run_id"] == _run_id() @@ -136,6 +137,35 @@ def test_state_hub_progress_sink_posts_compact_probe_summary(monkeypatch) -> Non assert "token=secret" not in serialized +def test_state_hub_progress_prefers_workplan_id(monkeypatch) -> None: + posts: list[dict[str, Any]] = [] + + def fake_get(url: str, **kwargs: Any) -> DummyResponse: + return DummyResponse([]) + + def fake_post(url: str, **kwargs: Any) -> DummyResponse: + posts.append({"url": url, **kwargs}) + return DummyResponse({"id": "progress-2"}) + + monkeypatch.setattr(httpx, "get", fake_get) + monkeypatch.setattr(httpx, "post", fake_post) + + persist_ops_inventory_evidence( + _payload([ + { + "type": "state-hub-progress", + "state_hub_url": "http://state-hub.test", + "event_type": "ops_inventory_probe", + "workplan_id": "workplan-1", + } + ]) + ) + + body = posts[0]["json"] + assert body["workplan_id"] == "workplan-1" + assert body["workstream_id"] == "workplan-1" + + def test_core_hub_stabilization_sink_posts_progress(monkeypatch) -> None: posts: list[dict[str, Any]] = [] diff --git a/tests/test_report_sinks.py b/tests/test_report_sinks.py index 5c882c9..e2c77fa 100644 --- a/tests/test_report_sinks.py +++ b/tests/test_report_sinks.py @@ -116,6 +116,7 @@ def test_state_hub_progress_sink_posts(monkeypatch) -> None: } ] assert posts[0]["url"] == "http://state-hub.test/progress/" + assert posts[0]["json"]["workplan_id"] == "workstream-1" assert posts[0]["json"]["workstream_id"] == "workstream-1" assert posts[0]["json"]["detail"]["activity_core_run_id"] == payload_run_id() assert posts[0]["json"]["detail"]["output_validated"] is True diff --git a/tests/test_schedule_health.py b/tests/test_schedule_health.py index 33f3715..9df8d3e 100644 --- a/tests/test_schedule_health.py +++ b/tests/test_schedule_health.py @@ -4,7 +4,9 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone -from activity_core.schedule_health import evaluate_schedule_health +import httpx + +from activity_core.schedule_health import evaluate_schedule_health, post_missed_fire_alert NOW = datetime(2026, 6, 23, 12, 0, tzinfo=timezone.utc) @@ -70,6 +72,35 @@ def test_no_fire_recorded_for_due_schedule_is_unhealthy() -> None: assert "no recorded fire" in health.reasons[0] +def test_post_missed_fire_alert_dual_writes_workplan_scope(monkeypatch) -> None: + posts: list[dict] = [] + + class _Resp: + def raise_for_status(self) -> None: ... + def json(self) -> dict[str, str]: + return {"id": "progress-1"} + + monkeypatch.setattr(httpx, "post", lambda url, **kwargs: posts.append(kwargs) or _Resp()) + + health = evaluate_schedule_health( + activity_id="daily", + missed_catchup_window=1, + last_fired_at=NOW - timedelta(days=2), + now=NOW, + expected_interval=timedelta(days=1), + ) + result = post_missed_fire_alert( + health, + state_hub_url="http://state-hub.test", + workplan_id="wp-123", + ) + + assert result["status"] == "posted" + body = posts[0]["json"] + assert body["workplan_id"] == "wp-123" + assert body["workstream_id"] == "wp-123" + + def test_no_interval_and_no_fire_is_not_flagged() -> None: # Without an expected interval we cannot assert a miss from absence alone. health = evaluate_schedule_health( diff --git a/tests/test_state_hub_context_resolver.py b/tests/test_state_hub_context_resolver.py index 951fd0d..0defda9 100644 --- a/tests/test_state_hub_context_resolver.py +++ b/tests/test_state_hub_context_resolver.py @@ -751,3 +751,5 @@ def test_phase5_stabilization_check_passes(monkeypatch) -> None: assert result["overall_pass"] is True assert result["skipped"] is False assert result["checks"]["totals"]["pass"] is True + assert result["checks"]["totals"]["workplans"] == 640 + assert result["checks"]["totals"]["workstreams"] == 640 diff --git a/tests/test_state_hub_write.py b/tests/test_state_hub_write.py index 207eaf9..bc89b77 100644 --- a/tests/test_state_hub_write.py +++ b/tests/test_state_hub_write.py @@ -8,6 +8,7 @@ import pytest from activity_core import report_sinks from activity_core.state_hub_write import ( IDEMPOTENCY_HEADER, + apply_progress_scope_fields, idempotency_headers, idempotency_key, ) @@ -33,6 +34,24 @@ def test_headers_carry_the_key() -> None: assert headers == {IDEMPOTENCY_HEADER: "run1:i:e"} +def test_apply_progress_scope_prefers_workplan_id() -> None: + body: dict[str, str] = {} + apply_progress_scope_fields( + body, + {"workplan_id": "wp-1", "workstream_id": "ws-legacy", "topic_id": "topic-1"}, + ) + assert body["workplan_id"] == "wp-1" + assert body["workstream_id"] == "wp-1" + assert body["topic_id"] == "topic-1" + + +def test_apply_progress_scope_dual_writes_legacy_workstream_id() -> None: + body: dict[str, str] = {} + apply_progress_scope_fields(body, {"workstream_id": "ws-1"}) + assert body["workplan_id"] == "ws-1" + assert body["workstream_id"] == "ws-1" + + def test_distinct_identities_get_distinct_keys() -> None: assert idempotency_key("r", "i", "daily_triage") != idempotency_key( "r", "i", "schedule_miss"