From 70e3154b05f36f345166f42ecbdf7f46709c340b Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 7 Jul 2026 00:48:36 +0200 Subject: [PATCH] feat: Phase 5 stabilization schedules and custodian path URIs Add phase5_stabilization_check State Hub resolver with progress evidence sinks, schedule projections for daily and closeout checks, custodian:// and activity-core:// runtime path resolution, and Railiance mounts under /var/custodian. --- k8s/railiance/20-runtime.yaml | 99 ++++++- k8s/railiance/21-custodian-runtime.yaml | 270 ++++++++++++++++++ src/activity_core/activities.py | 9 +- .../context_resolvers/state_hub.py | 151 ++++++++++ src/activity_core/ops_evidence_sinks.py | 41 ++- src/activity_core/report_sinks.py | 15 +- src/activity_core/rules/executor.py | 4 +- src/activity_core/runtime_paths.py | 37 +++ tests/test_runtime_paths.py | 17 ++ tests/test_state_hub_context_resolver.py | 47 +++ 10 files changed, 672 insertions(+), 18 deletions(-) create mode 100644 k8s/railiance/21-custodian-runtime.yaml create mode 100644 src/activity_core/runtime_paths.py create mode 100644 tests/test_runtime_paths.py diff --git a/k8s/railiance/20-runtime.yaml b/k8s/railiance/20-runtime.yaml index 4650477..091f428 100644 --- a/k8s/railiance/20-runtime.yaml +++ b/k8s/railiance/20-runtime.yaml @@ -17,6 +17,8 @@ data: ISSUE_CORE_URL: http://actcore-issue-core-bridge.activity-core.svc.cluster.local:8765 ISSUE_SINK_TYPE: "rest" ACTIVITY_DEFINITION_DIRS: /etc/activity-core/external-definitions + CUSTODIAN_REPO_ROOT: /var/custodian + ACTIVITY_CORE_ROOT: /etc/activity-core OPS_INVENTORY_PATH: /etc/activity-core/ops/service-inventory.yml INTER_HUB_URL: "" OPS_HUB_WIDGET_MAPPING: "" @@ -55,7 +57,7 @@ data: - type: static bind_to: context.prompt_path config: - value: /home/worsch/the-custodian/runtime/prompts/daily_statehub_wsgi_triage.md + value: custodian://runtime/prompts/daily_statehub_wsgi_triage.md - type: state-hub query: daily_triage_digest params: @@ -102,7 +104,7 @@ data: {context.daily_triage_digest} Return only JSON matching - `/etc/activity-core/schemas/daily-triage-report.json`. Emit the "summary" + `activity-core://schemas/daily-triage-report.json`. Emit the "summary" field first, then inside the "recommendations" array write one complete recommendation JSON object per line (NDJSON-style per-item framing) so each item can be recovered independently if the output is truncated. Do @@ -127,11 +129,11 @@ data: } ] } - output_schema: /etc/activity-core/schemas/daily-triage-report.json + output_schema: activity-core://schemas/daily-triage-report.json review_required: false report_sinks: - type: working-memory - path: /home/worsch/the-custodian/memory/working + path: custodian://memory/working timezone: Europe/Berlin filename_template: "daily-triage-{date}-{run_id_short}.md" - type: state-hub-progress @@ -201,6 +203,87 @@ data: Kubernetes projection of the Custodian-owned definition in `/home/worsch/the-custodian/activity-definitions/state-hub-consistency-sweep.md`. + phase5-stabilization-daily.md: | + --- + id: "f3a8c2e1-9b4d-4a6f-8e2d-1c5b7a9e3f04" + name: "Phase 5 Stabilization Daily Check" + type: activity-definition + version: "1.0" + enabled: true + owner: custodian + governance: custodian + status: active + created: "2026-07-06" + trigger: + type: cron + cron_expression: "0 9 * * *" + timezone: Europe/Berlin + misfire_policy: skip + context_sources: + - type: state-hub + query: phase5_stabilization_check + required: true + params: + source: activity-core + closeout: false + window_start: "2026-07-06T17:35:00+00:00" + window_end: "2026-07-09T17:35:00+00:00" + baseline: + workstreams: 640 + tasks: 4002 + topics: 14 + sweep_limit: 6 + triage_max_age_hours: 36 + evidence_sinks: + - type: state-hub-progress + event_type: phase5_stabilization_check + author: activity-core + workstream_id: 8a828444-dd49-4d7b-a2d1-9952b5bc929d + task_id: e91db8d0-973d-4a31-b3c2-ca37fd002ec7 + bind_to: context.phase5_stabilization_check + --- + + # ActivityDefinition: Phase 5 Stabilization Daily Check + phase5-stabilization-closeout.md: | + --- + id: "e7d2b5a8-4c1f-4e9a-b6d3-8f2a1c4e6b09" + name: "Phase 5 Stabilization Closeout Check" + type: activity-definition + version: "1.0" + enabled: true + owner: custodian + governance: custodian + status: active + created: "2026-07-06" + trigger: + type: scheduled + at: "2026-07-09T17:35:00+00:00" + timezone: UTC + context_sources: + - type: state-hub + query: phase5_stabilization_check + required: true + params: + source: activity-core + closeout: true + window_start: "2026-07-06T17:35:00+00:00" + window_end: "2026-07-09T17:35:00+00:00" + baseline: + workstreams: 640 + tasks: 4002 + topics: 14 + sweep_limit: 6 + triage_max_age_hours: 36 + evidence_sinks: + - type: state-hub-progress + event_type: phase5_stabilization_closeout + author: activity-core + workstream_id: 8a828444-dd49-4d7b-a2d1-9952b5bc929d + task_id: e91db8d0-973d-4a31-b3c2-ca37fd002ec7 + bind_to: context.phase5_stabilization_check + --- + + # ActivityDefinition: Phase 5 Stabilization Closeout Check ops-service-inventory-probes.md: | --- id: "40d15a87-7ff6-4d8e-992c-37df15f95110" @@ -953,7 +1036,10 @@ spec: mountPath: /etc/activity-core/ops readOnly: true - name: working-memory - mountPath: /home/worsch/the-custodian/memory/working + mountPath: /var/custodian/memory/working + - name: custodian-runtime + mountPath: /var/custodian/runtime/prompts + readOnly: true volumes: - name: external-activity-definitions configMap: @@ -967,6 +1053,9 @@ spec: - name: working-memory persistentVolumeClaim: claimName: actcore-working-memory + - name: custodian-runtime + configMap: + name: actcore-custodian-runtime --- apiVersion: apps/v1 kind: Deployment diff --git a/k8s/railiance/21-custodian-runtime.yaml b/k8s/railiance/21-custodian-runtime.yaml new file mode 100644 index 0000000..8d73ed0 --- /dev/null +++ b/k8s/railiance/21-custodian-runtime.yaml @@ -0,0 +1,270 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: actcore-custodian-runtime + namespace: activity-core + labels: + app.kubernetes.io/name: activity-core + app.kubernetes.io/part-of: activity-core +data: + daily_statehub_wsgi_triage.md: | + --- + id: daily-statehub-wsjf-triage + type: runtime-prompt + owner: custodian + status: active + created: "2026-05-17" + updated: "2026-06-04" + related_workplan: CUST-WP-0044 + --- + + # Daily State Hub WSJF Triage Prompt + + ## Purpose + + Run a daily Custodian review of State Hub state and produce a short, + reviewable recommendation report. The review is a focus surface, not an + execution loop: it recommends what to work next, what to revisit, what to park, + and what needs human or cross-agent attention. + + Do not create a new scheduler for this loop. The current runner is activity-core + using the ActivityDefinition in + `/home/worsch/the-custodian/activity-definitions/daily-statehub-wsjf-triage.md`. + Do not re-enable a Codex app fallback unless activity-core is deliberately + disabled and that operator decision is recorded, so there is exactly one daily + runner. + + ## Operating Rules + + - Read first, then decide. Do not edit workplans, canon, or task status during + a daily review unless a later human request explicitly asks for an apply step. + - Prefer existing State Hub read surfaces over ad hoc scans: + `/state/summary`, `/state/next_steps`, `/workstreams/workplan-index`, + `/messages/`, and `.custodian-brief.md`. + - Use local workplan files only to enrich the State Hub snapshot with task + wording, stale frontmatter, and missing file-backed state. + - Treat scores as prioritization aids, not truth. Every WSJF row needs a + confidence label. + - Include explicit WSJF component scores in executable JSON output so the + working-memory note remains auditable without re-running the model. + - Items involving money, legal status, secrets, security posture, external + publication, or external reputation must be reported as `needs-human` unless + a current explicit approval already exists. + + ## Inputs + + Collect these inputs in order. If a live input fails, continue from the last + available fallback and say so in the report. + + 1. `/home/worsch/the-custodian/.custodian-brief.md` + 2. `GET http://127.0.0.1:8000/state/summary` + 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 + 7. Read-only `git status --short` for repos named in top recommendations + + Optional enrichment: + + - `GET /tasks/?workstream_id=` for a top-ranked workstream + - `GET /progress/?workstream_id=&limit=5` for staleness confidence + - State Hub domain summaries through MCP when available + + ## Candidate Set + + Build the candidate list from: + + - all open workstreams in `state_summary.open_workstreams` + - all derived `state_summary.next_steps` + - blocked tasks and blocking decisions from the summary + - high-priority file-backed workplans surfaced by `workplan-index` + - workstreams with suspicious structure, such as zero parsed tasks or stale + active plans + + Keep the scored table compact. Score at most 15 candidates internally and + report the top 10. + + ## Loose-End Detection + + Flag a candidate when one or more signals apply: + + | Signal | Default threshold | Recommended action | + |--------|-------------------|--------------------| + | 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` | + | 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` | + + Use these as prompts for judgment, not as automatic status changes. + + ## WSJF Procedure + + Score every reported candidate with: + + ```text + WSJF = (strategic_value + time_criticality + risk_reduction + opportunity_enablement) / job_size + ``` + + Use integer scores from 1 to 5. + + ### Strategic Value + + Score higher when the work aligns with: + + - current Custodian/State Hub operating reliability + - 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 + + ### Time Criticality + + Score higher when waiting increases: + + - operational risk or data-loss risk + - number of blocked downstream plans + - context decay for partially completed work + - queue pressure from accumulating open tasks + + ### Risk Reduction + + Score higher when the work reduces: + + - security, identity, backup, restore, deployment, or consistency risk + - false dashboard state or stale State Hub records + - manual recovery burden + - ambiguity around human approvals + + ### Opportunity Enablement + + Score higher when the work unlocks: + + - multiple downstream workstreams + - agent autonomy or fewer repeated setup steps + - reliable cross-repo operation + - future daily triage quality + + ### Job Size + + Score lower numbers for easier jobs: + + | Score | Meaning | + |-------|---------| + | 1 | tiny, clear, can likely close in one focused session | + | 2 | small, local, low uncertainty | + | 3 | moderate, several files or one external dependency | + | 4 | large, multi-repo, needs coordination or testing | + | 5 | broad/uncertain, likely should be split before execution | + + ### Confidence + + Use: + + - `high` when State Hub summary, workplan file, and recent progress agree + - `medium` when the score relies on summary data plus one corroborating source + - `low` when the workplan file is missing, stale, blocked by dirty repo state, + or based mostly on inference + + ## Recommendation Actions + + | Action | Meaning | Agent may proceed when | Human gate required when | + |--------|---------|------------------------|--------------------------| + | `work-next` | Best next executable task | scope is local, reversible, and inside an existing approved workplan | it touches money, legal, secrets, public reputation, or external commitments | + | `revisit` | Re-read and refresh before execution | the plan is stale, ambiguous, blocked, or context has moved | revisiting changes purpose, scope, owner, or approval posture | + | `split` | Break an oversized workplan into smaller plans | split is file-backed and preserves provenance | split would drop scope, change priorities, or alter commitments | + | `park` | Move out of active focus | plan is clearly not current and parking is proposed only | actually changing status to backlog/archive needs review | + | `close-out` | Finish closure review and mark done when appropriate | remaining tasks are truly done/cancelled/carry-forwarded | tasks are ambiguous, cancelled for policy reasons, or external effects are involved | + | `needs-human` | Human decision or approval needed | never auto-resolve; report the ask crisply | always | + | `needs-cross-agent` | Another repo/agent is the right owner | send/prepare a coordination message or task only when requested | when ownership or priority is uncertain | + | `needs-consistency-sync` | File/DB/index state should be reconciled | running read/check/fix consistency is already allowed by repo protocol | sync would overwrite work or local repo is behind remote | + + ## Report Template + + Write the daily note in this shape: + + ```markdown + --- + id: daily-statehub-triage-YYYY-MM-DD + type: working-memory + created: "YYYY-MM-DD" + related_workplan: CUST-WP-0044 + source: daily-state-hub-wsjf-triage + --- + + # Daily State Hub WSJF Triage - YYYY-MM-DD + + ## Snapshot + + - Generated at: + - Workstreams: active, blocked, finished + - Tasks: todo, in progress, blocked + - Decisions: open, escalated + - Inbox: unread hub messages + - Input health: + + ## Top Recommendations + + | Rank | Action | Candidate | WSJF | Confidence | Why now | + |------|--------|-----------|------|------------|---------| + | 1 | work-next | | 0.0 | high | | + + ## Loose Ends + + | Action | Candidate | Signal | Suggested follow-up | + |--------|-----------|--------|---------------------| + | revisit | | | | + + ## Human Or Cross-Agent Attention + + - `needs-human`: - + - `needs-cross-agent`: - + + ## Next Custodian Session + + 1. + 2. + 3. + + ## Progress Event Summary + + + ``` + + ## Progress Event + + 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` + - `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 + + Do not mark tasks done from the daily run itself. Task status changes belong to + the implementation session that applies a recommendation. + + ## Executable JSON Shape + + When the activity-core runner asks for JSON only, return the same content in + the schema at `/home/worsch/the-custodian/schemas/daily-triage-report.json`. + Each recommendation must include: + + - `rank` + - `candidate` + - `action` + - `why` + - `confidence` + - `wsjf.score` + - `wsjf.strategic_value` + - `wsjf.time_criticality` + - `wsjf.risk_reduction` + - `wsjf.opportunity_enablement` + - `wsjf.job_size` +kind: ConfigMap +metadata: + name: actcore-custodian-runtime diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 5585ab4..0d30726 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -156,7 +156,14 @@ async def resolve_context( bind_key = raw_bind.removeprefix("context.") if raw_bind.startswith("context.") else raw_bind if source_type == "static": - snapshot[bind_key] = source.get("config", {}).get("value") + value = source.get("config", {}).get("value") + if isinstance(value, str) and ( + value.startswith("custodian://") or value.startswith("activity-core://") + ): + from activity_core.runtime_paths import resolve_runtime_path + + value = str(resolve_runtime_path(value)) + snapshot[bind_key] = value continue resolver_cls = CONTEXT_RESOLVER_REGISTRY.get(source_type) diff --git a/src/activity_core/context_resolvers/state_hub.py b/src/activity_core/context_resolvers/state_hub.py index 3b7bd78..0ebd140 100644 --- a/src/activity_core/context_resolvers/state_hub.py +++ b/src/activity_core/context_resolvers/state_hub.py @@ -13,6 +13,7 @@ Supported queries: - daily_triage_digest: curated scalar JSON digest for daily WSJF triage - recently_on_scope_hourly: POST {STATE_HUB_URL}/recently-on-scope/hourly - consistency_sweep_remote_all: POST {STATE_HUB_URL}/consistency/sweep/remote-all + - phase5_stabilization_check: hub-visible Phase 5 stabilization gates No caching — state hub data is live operational state and must not be stale within a single workflow run. @@ -134,9 +135,159 @@ class StateHubContextResolver(ContextResolver): timeout=_SWEEP_TIMEOUT_SECONDS, ) return _validate_consistency_sweep_remote_all(result) + if query == "phase5_stabilization_check": + return _phase5_stabilization_check(params) return {} +def _parse_iso_datetime(raw: Any) -> datetime | None: + if not raw: + return None + text = str(raw).replace("Z", "+00:00") + try: + dt = datetime.fromisoformat(text) + except ValueError: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]: + closeout = bool(params.get("closeout")) + now = _utc_now() + window_start = _parse_iso_datetime( + params.get("window_start", "2026-07-06T17:35:00+00:00") + ) + window_end = _parse_iso_datetime( + params.get("window_end", "2026-07-09T17:35:00+00:00") + ) + within_window = bool( + window_start + and window_end + and window_start <= now <= window_end + ) + + if not within_window and not closeout: + return { + "skipped": True, + "reason": "outside_stabilization_window", + "closeout": closeout, + "overall_pass": True, + "window": { + "start": window_start.isoformat() if window_start else None, + "end": window_end.isoformat() if window_end else None, + "within": within_window, + }, + "checks": {}, + } + + baseline = params.get("baseline") or {} + expected_workstreams = int(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) + triage_max_age_hours = _bounded_int( + params.get("triage_max_age_hours", 36), + default=36, + minimum=1, + maximum=168, + ) + + health = _fetch_json("/state/health") + health_pass = isinstance(health, dict) and health.get("status") == "ok" + + summary = _fetch_json("/state/summary") + totals = (summary or {}).get("totals") or {} + ws_total = int((totals.get("workstreams") or {}).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 + and task_total == expected_tasks + and topic_total == expected_topics + ) + + sweep_items = _fetch_json( + "/progress/", + {"event_type": "consistency_sweep_remote_all", "limit": sweep_limit}, + ) + sweep_rows = sweep_items if isinstance(sweep_items, list) else [] + sweep_failures = 0 + sweep_missing = 0 + for item in sweep_rows: + detail = item.get("detail") or {} + if detail.get("exit_code") != 0 or detail.get("automation_error"): + sweep_failures += 1 + missing = detail.get("skipped_missing") or [] + if isinstance(missing, list) and missing: + sweep_missing += len(missing) + sweeps_pass = bool(sweep_rows) and sweep_failures == 0 and sweep_missing == 0 + + triage_items = _fetch_json( + "/progress/", + {"event_type": "daily_triage", "limit": 1}, + ) + triage_row = triage_items[0] if isinstance(triage_items, list) and triage_items else None + triage_at = _parse_iso_datetime(triage_row.get("created_at") if triage_row else None) + triage_age_hours = ( + (now - triage_at).total_seconds() / 3600 if triage_at else None + ) + triage_pass = bool( + triage_row + and triage_age_hours is not None + and triage_age_hours <= triage_max_age_hours + ) + + checks = { + "health": {"pass": health_pass, "status": health.get("status") if isinstance(health, dict) else None}, + "totals": { + "pass": totals_pass, + "workstreams": ws_total, + "tasks": task_total, + "topics": topic_total, + "expected": { + "workstreams": expected_workstreams, + "tasks": expected_tasks, + "topics": expected_topics, + }, + }, + "sweeps": { + "pass": sweeps_pass, + "sampled": len(sweep_rows), + "failures": sweep_failures, + "missing": sweep_missing, + }, + "daily_triage": { + "pass": triage_pass, + "last_at": triage_at.isoformat() if triage_at else None, + "age_hours": round(triage_age_hours, 2) if triage_age_hours is not None else None, + "max_age_hours": triage_max_age_hours, + }, + } + overall_pass = all(section.get("pass") for section in checks.values()) + result = { + "skipped": False, + "closeout": closeout, + "overall_pass": overall_pass, + "operator_signoff_needed": closeout and overall_pass, + "window": { + "start": window_start.isoformat() if window_start else None, + "end": window_end.isoformat() if window_end else None, + "within": within_window, + }, + "checks": checks, + "source": str(params.get("source") or "activity-core"), + } + if closeout and not overall_pass: + raise RuntimeError(f"phase5 stabilization closeout failed: {result}") + return result + + CONTEXT_RESOLVER_REGISTRY["state-hub"] = StateHubContextResolver diff --git a/src/activity_core/ops_evidence_sinks.py b/src/activity_core/ops_evidence_sinks.py index 8209084..a78901b 100644 --- a/src/activity_core/ops_evidence_sinks.py +++ b/src/activity_core/ops_evidence_sinks.py @@ -33,7 +33,10 @@ def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, An """ results: list[dict[str, Any]] = [] for source in payload.get("context_sources", []): - if not isinstance(source, dict) or source.get("type") != "ops-inventory": + if not isinstance(source, dict): + continue + source_type = source.get("type") + if source_type not in {"ops-inventory", "state-hub"}: continue params = source.get("params") or {} @@ -43,6 +46,14 @@ def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, An bind_key = _context_bind_key(source) probe_result = (payload.get("context") or {}).get(bind_key) + if isinstance(probe_result, dict) and probe_result.get("skipped"): + results.append({ + "type": "state-hub-progress", + "status": "skipped", + "reason": probe_result.get("reason", "skipped"), + "context_key": bind_key, + }) + continue if not isinstance(probe_result, dict): results.extend( { @@ -112,16 +123,23 @@ def _post_state_hub_progress( "context_key": context_key, } - compact = _compact_probe_result(probe_result) + if probe_result.get("checks") is not None: + compact = probe_result + summary = _phase5_summary_text(probe_result) + source_type = "state-hub" + else: + compact = _compact_probe_result(probe_result) + summary = _summary_text(compact.get("summary", {})) + source_type = "ops-inventory" body: dict[str, Any] = { "event_type": event_type, "author": sink.get("author", "activity-core"), - "summary": _summary_text(compact.get("summary", {})), + "summary": summary, "detail": { "activity_id": payload.get("activity_id"), "activity_core_run_id": run_id, "scheduled_for": payload.get("scheduled_for"), - "source_type": "ops-inventory", + "source_type": source_type, "context_key": context_key, "idempotency_key": idempotency_key, "probe": compact, @@ -485,6 +503,21 @@ def _compact_access_path(access_path: dict[str, Any]) -> dict[str, Any]: } +def _phase5_summary_text(result: dict[str, Any]) -> str: + checks = result.get("checks") or {} + totals = checks.get("totals") or {} + sweeps = checks.get("sweeps") or {} + triage = checks.get("daily_triage") or {} + status = "pass" if result.get("overall_pass") else "fail" + mode = "closeout" if result.get("closeout") else "daily" + return ( + f"Phase 5 stabilization {mode}: {status}; " + f"totals {totals.get('workstreams', '?')}/{totals.get('tasks', '?')}/" + f"{totals.get('topics', '?')}; sweeps sampled={sweeps.get('sampled', 0)}; " + f"daily_triage age_h={triage.get('age_hours', '?')}" + ) + + def _summary_text(summary: dict[str, Any]) -> str: return ( "Ops inventory probe: " diff --git a/src/activity_core/report_sinks.py b/src/activity_core/report_sinks.py index 8fd0cdb..2c01e70 100644 --- a/src/activity_core/report_sinks.py +++ b/src/activity_core/report_sinks.py @@ -11,14 +11,10 @@ from zoneinfo import ZoneInfo import httpx +from activity_core.runtime_paths import custodian_repo_root, resolve_runtime_path from activity_core.state_hub_write import idempotency_headers _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000" -_THE_CUSTODIAN_ROOT = Path("/home/worsch/the-custodian") -_FORBIDDEN_CUSTODIAN_ROOTS = ( - _THE_CUSTODIAN_ROOT / "canon", - _THE_CUSTODIAN_ROOT / "workplans", -) def persist_reports(payload: dict[str, Any]) -> list[dict[str, Any]]: @@ -66,7 +62,7 @@ def _write_working_memory( report_entry: dict[str, Any], sink: dict[str, Any], ) -> dict[str, Any]: - directory = Path(sink.get("path", "")).expanduser() + directory = resolve_runtime_path(str(sink.get("path", ""))) if not directory: raise ValueError("working-memory sink requires path") @@ -257,7 +253,12 @@ def _local_date(scheduled_for: str | None, timezone_name: str) -> str: def _assert_allowed_output_path(path: Path) -> None: - for forbidden in _FORBIDDEN_CUSTODIAN_ROOTS: + root = custodian_repo_root() + forbidden_roots = ( + root / "canon", + root / "workplans", + ) + for forbidden in forbidden_roots: try: path.relative_to(forbidden) except ValueError: diff --git a/src/activity_core/rules/executor.py b/src/activity_core/rules/executor.py index 5a56a12..10ec521 100644 --- a/src/activity_core/rules/executor.py +++ b/src/activity_core/rules/executor.py @@ -799,7 +799,9 @@ def _load_output_schema(schema_path: str) -> dict[str, Any] | None: if not schema_path: return None - path = Path(schema_path) + from activity_core.runtime_paths import resolve_runtime_path + + path = resolve_runtime_path(schema_path) if not path.exists(): return None diff --git a/src/activity_core/runtime_paths.py b/src/activity_core/runtime_paths.py new file mode 100644 index 0000000..7e7a03f --- /dev/null +++ b/src/activity_core/runtime_paths.py @@ -0,0 +1,37 @@ +"""Resolve repo-relative runtime paths for Custodian-owned assets.""" + +from __future__ import annotations + +import os +from pathlib import Path + +_DEFAULT_CUSTODIAN_ROOT = Path("/home/worsch/the-custodian") +_DEFAULT_ACTIVITY_CORE_ROOT = Path("/etc/activity-core") + +_CUSTODIAN_SCHEME = "custodian://" +_ACTIVITY_CORE_SCHEME = "activity-core://" + + +def custodian_repo_root() -> Path: + raw = os.environ.get("CUSTODIAN_REPO_ROOT", "").strip() + return Path(raw).expanduser() if raw else _DEFAULT_CUSTODIAN_ROOT + + +def activity_core_root() -> Path: + raw = os.environ.get("ACTIVITY_CORE_ROOT", "").strip() + return Path(raw).expanduser() if raw else _DEFAULT_ACTIVITY_CORE_ROOT + + +def resolve_runtime_path(raw_path: str) -> Path: + """Map custodian:// and activity-core:// URIs to mounted runtime paths.""" + value = str(raw_path or "").strip() + if not value: + return Path(value) + + if value.startswith(_CUSTODIAN_SCHEME): + return (custodian_repo_root() / value.removeprefix(_CUSTODIAN_SCHEME)).resolve() + + if value.startswith(_ACTIVITY_CORE_SCHEME): + return (activity_core_root() / value.removeprefix(_ACTIVITY_CORE_SCHEME)).resolve() + + return Path(value).expanduser() \ No newline at end of file diff --git a/tests/test_runtime_paths.py b/tests/test_runtime_paths.py new file mode 100644 index 0000000..8c4c731 --- /dev/null +++ b/tests/test_runtime_paths.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from activity_core.runtime_paths import resolve_runtime_path + + +def test_resolve_custodian_uri(monkeypatch) -> None: + monkeypatch.setenv("CUSTODIAN_REPO_ROOT", "/var/custodian") + assert resolve_runtime_path("custodian://memory/working") == ( + __import__("pathlib").Path("/var/custodian/memory/working") + ) + + +def test_resolve_activity_core_uri(monkeypatch) -> None: + monkeypatch.setenv("ACTIVITY_CORE_ROOT", "/etc/activity-core") + assert resolve_runtime_path("activity-core://schemas/daily-triage-report.json") == ( + __import__("pathlib").Path("/etc/activity-core/schemas/daily-triage-report.json") + ) \ No newline at end of file diff --git a/tests/test_state_hub_context_resolver.py b/tests/test_state_hub_context_resolver.py index 0ff5367..39b0b6b 100644 --- a/tests/test_state_hub_context_resolver.py +++ b/tests/test_state_hub_context_resolver.py @@ -607,3 +607,50 @@ def test_daily_triage_digest_is_curated_scalar_json(monkeypatch) -> None: "code_score_high_gain_high_effort_candidates" ) assert digest["ranked_suggestions"][0]["origin_ref"] == "issue-core-ingestion-api-key" + + +def test_phase5_stabilization_check_passes(monkeypatch) -> None: + payloads = { + "/state/health": {"status": "ok", "db": "connected"}, + "/state/summary": { + "totals": { + "workstreams": {"total": 640}, + "tasks": {"total": 4002}, + "topics": {"total": 14}, + } + }, + "/progress/": [], + } + + def fake_get(url: str, **kwargs: Any) -> DummyResponse: + path = url.replace("http://state-hub.test", "") + params = kwargs.get("params") or {} + if path == "/progress/": + if params.get("event_type") == "consistency_sweep_remote_all": + return DummyResponse([ + { + "created_at": "2026-07-06T20:00:00+00:00", + "detail": {"exit_code": 0, "skipped_missing": []}, + } + ]) + if params.get("event_type") == "daily_triage": + return DummyResponse([ + {"created_at": "2026-07-06T19:00:00+00:00", "summary": "ok"} + ]) + return DummyResponse(payloads.get(path, {})) + + from datetime import datetime, timezone + + fixed_now = datetime(2026, 7, 7, 10, 0, tzinfo=timezone.utc) + monkeypatch.setenv("STATE_HUB_URL", "http://state-hub.test") + monkeypatch.setattr(httpx, "get", fake_get) + monkeypatch.setattr( + "activity_core.context_resolvers.state_hub._utc_now", + lambda: fixed_now, + ) + + result = StateHubContextResolver().resolve("phase5_stabilization_check", None, {}) + + assert result["overall_pass"] is True + assert result["skipped"] is False + assert result["checks"]["totals"]["pass"] is True