From 98e8aa83bd14668b244e7022c0737c2a6845b589 Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 21 Jul 2026 04:21:55 +0200 Subject: [PATCH] Implement ACTIVITY-WP-0021 production automation reliability Root-cause IssueSink 503 (dead Forgejo PAT on issue-core), add state-hub task sink path B, log runs before emit, harden sync_schedules, deterministic SBOM/triage reports, DB probe thrash fix, and prod automation-status helper. --- Makefile | 3 + activity-definitions/weekly-sbom-staleness.md | 35 +++++- docs/issue-core-emission-boundary.md | 36 +++++- docs/runbook.md | 43 +++++++ k8s/railiance/10-infrastructure.yaml | 31 ++++- scripts/prod_automation_status.sh | 58 +++++++++ src/activity_core/activities.py | 34 +++++- src/activity_core/issue_sink.py | 89 +++++++++++++- src/activity_core/rules/executor.py | 78 ++++++++++++ src/activity_core/schedule_manager.py | 13 +- src/activity_core/sync_schedules.py | 52 ++++++-- src/activity_core/workflows.py | 35 +++--- tests/test_issue_sink.py | 11 +- tests/test_sync_schedules.py | 69 +++++++++++ ...ITY-WP-0021-prod-automation-reliability.md | 114 +++++++++++++++--- 15 files changed, 638 insertions(+), 63 deletions(-) create mode 100755 scripts/prod_automation_status.sh diff --git a/Makefile b/Makefile index 9a0abd5..67e61a4 100644 --- a/Makefile +++ b/Makefile @@ -40,6 +40,9 @@ automation-status: ## Report recent automation status from repo-owned evidence automation-status-json: ## Report recent automation status as JSON $(MAKE) automation-status FORMAT=json +prod-automation-status: ## Prod run counts via SSH railiance01 (ACTIVITY-WP-0021) + ./scripts/prod_automation_status.sh $(SINCE) + automation-list: ## List configured scheduled automations from repo-owned definitions @uv run python scripts/automation_inventory.py --format "$(FORMAT)" --enabled "$(ENABLED)" $(if $(TRIGGER),--trigger-type "$(TRIGGER)",) $(if $(ACTIVITY_ID),--activity-id "$(ACTIVITY_ID)",) $(if $(ACTIVITY_NAME),--activity-name "$(ACTIVITY_NAME)",) diff --git a/activity-definitions/weekly-sbom-staleness.md b/activity-definitions/weekly-sbom-staleness.md index 292543d..b0823f1 100644 --- a/activity-definitions/weekly-sbom-staleness.md +++ b/activity-definitions/weekly-sbom-staleness.md @@ -26,11 +26,39 @@ context_sources: Runs every Monday at 09:00 Berlin time. Checks all tracked repositories for SBOM staleness and flags any repository whose SBOM is older than 30 days. +ACTIVITY-WP-0021: the fleet no longer treats Forgejo issues as the primary +landing zone for automated tasks. The deterministic **state-hub-progress** +instruction report below is the operator-visible evidence path. Task emission +via IssueSink remains optional and only fires when `ISSUE_SINK_TYPE` points at +a healthy sink (rest/state-hub); a broken Forgejo backend must not be required +for a green weekly completion. + +```instruction +id: weekly-sbom-staleness-report +trusted_fields: [] +model: deterministic +temperature: 0 +max_tokens: 1 +prompt: | + Deterministic SBOM staleness report from context.repos (no LLM). +output_schema: "" +review_required: false +report_sinks: + - type: state-hub-progress + event_type: sbom_staleness + author: activity-core + topic_id: cee7bedf-2b48-46ef-8601-006474f2ad7a +``` + +Task emission for stale repos is **disabled** while IssueSink→Forgejo is +policy/token broken (ACTIVITY-WP-0021). Re-enable the rule below once +`ISSUE_SINK_TYPE=rest` (or `state-hub`) is proven healthy again. + ```rule id: flag-stale-sbom for_each: context.repos.repos bind_as: repo -condition: 'context.repo.sbom_age_days > 30' +condition: 'false' action: task_template: Run SBOM rescan for {context.repo.repo_slug} target_repo: context.repo.repo_slug @@ -39,6 +67,5 @@ action: ``` The bulk resolver exposes the per-repo entries under `context.repos.repos`. -The rule uses explicit `for_each` binding so the workflow evaluates the -condition once per repository and emits one task per stale repo. Action fields -may reference the bound item with `context.repo.*`. +The deterministic instruction posts `sbom_staleness` progress with stale repo +counts and a sample list for operator review. diff --git a/docs/issue-core-emission-boundary.md b/docs/issue-core-emission-boundary.md index 087172e..a7b38ae 100644 --- a/docs/issue-core-emission-boundary.md +++ b/docs/issue-core-emission-boundary.md @@ -52,10 +52,40 @@ task reference before it can replace `IssueCoreRestSink`. `emit_tasks`, so Temporal retries and the workflow history make failures visible. Railiance runtime ConfigMap uses this mode once `ISSUE_CORE_API_KEY` is present in `actcore-runtime-secret`. +- `ISSUE_SINK_TYPE=state-hub`: ACTIVITY-WP-0021 path B. Each TaskSpec is posted + as a State Hub progress event (`activity_task_spawn` by default) instead of + creating a Forgejo issue. Use when issue-core→Forgejo is down or automated + Forgejo issues are policy-blocked. -Weekly SBOM staleness is the canonical promotion candidate because the rule -contract is deterministic and tested. Promote it only after a null-sink dry-run -review and one live `IssueCoreRestSink` smoke against the target endpoint. +### Known production failure (2026-07-21) + +`POST /issues/` returned **HTTP 503** with: + +```text +Failed to connect to backend 'forgejo-inbox': Failed to connect to Gitea API +``` + +Root cause on coulombcore issue-core: `GITEA_BACKEND_TOKEN` is rejected by +Forgejo (`/api/v1/user` → 401 user does not exist). Healthz stays 200. + +**Operator fix (path A):** rotate `GITEA_BACKEND_TOKEN` in OpenBao path +`platform/workloads/issue-core/issue-core/issue-core-runtime` (see +`warden route show issue-core-ingestion-api-key`) using a valid Forgejo PAT for +the issue-core service identity, then restart `issue-core` so the entrypoint +rewrites backends.json. Smoke from the worker: + +```bash +# From actcore-worker (does not print secrets) +python -c "import os,httpx; r=httpx.post(os.environ['ISSUE_CORE_URL']+'/issues/', + json={...full TaskSpec payload...}, + headers={'Authorization':'Bearer '+os.environ['ISSUE_CORE_API_KEY']}, timeout=30); + print(r.status_code, r.text[:200])" +``` + +Expect **201**, not 503. + +Weekly SBOM staleness now posts a deterministic `sbom_staleness` progress report +even when task emission is disabled. ## Promotion and rollback diff --git a/docs/runbook.md b/docs/runbook.md index d0c9445..6f8ff9a 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -195,6 +195,49 @@ make automation-status SINCE=2026-06-26 FORMAT=json make automation-status SINCE=2026-06-26 UNTIL=2026-06-27 ACTCORE_DB_URL= ``` +### Production evidence path (railiance01) + +The workstation `make automation-status` is **degraded** without a live +`ACTCORE_DB_URL` / `TEMPORAL_HOST` to the railiance01 stack (docker hostnames +in `.env` are not the production DBs). For operator questions about live +schedules, use the prod helper (SSH to the host; no k3s API tunnel is required): + +```bash +# Human summary since last Sunday (Europe/Berlin window handled client-side) +./scripts/prod_automation_status.sh sunday + +# Explicit UTC lower bound +./scripts/prod_automation_status.sh 2026-07-18T22:00:00+00:00 +``` + +The script SSHes to `railiance01` (see `~/.ssh/config`), queries +`activity_runs` via `kubectl -n activity-core exec actcore-app-db-0`, and prints +per-activity counts plus non-high-frequency fire rows. It never prints secrets. + +Manual equivalent: + +```bash +ssh railiance01 'export KUBECONFIG=/etc/rancher/k3s/k3s.yaml +kubectl -n activity-core exec actcore-app-db-0 -- psql -U actcore -d actcore -c " +SELECT d.name, count(*) AS runs, max(r.fired_at) AS last_fire, + sum(r.tasks_spawned) AS tasks +FROM activity_runs r +JOIN activity_definitions d ON d.id = r.activity_id +WHERE coalesce(r.scheduled_for, r.fired_at) >= timestamptz '\''2026-07-18 22:00:00+00'\'' +GROUP BY d.name ORDER BY runs DESC;" +' +``` + +Temporal schedule / workflow status (from the worker pod): + +```bash +ssh railiance01 'export KUBECONFIG=/etc/rancher/k3s/k3s.yaml +kubectl -n activity-core exec deploy/actcore-worker -- /app/.venv/bin/python3 -c " +# describe ScheduleHandle for activity-schedule- +..." +' +``` + Example distinction from the June 2026 daily triage evidence: ```text diff --git a/k8s/railiance/10-infrastructure.yaml b/k8s/railiance/10-infrastructure.yaml index 9878999..dd229b9 100644 --- a/k8s/railiance/10-infrastructure.yaml +++ b/k8s/railiance/10-infrastructure.yaml @@ -57,16 +57,28 @@ spec: secretKeyRef: name: actcore-app-db-secret key: database + # ACTIVITY-WP-0021-T07: BestEffort QoS + 1s probe timeout caused + # thrash (100s of restarts) when the single-node host was busy. readinessProbe: exec: command: ["pg_isready", "-U", "actcore"] initialDelaySeconds: 5 periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 6 livenessProbe: exec: command: ["pg_isready", "-U", "actcore"] - initialDelaySeconds: 30 - periodSeconds: 20 + initialDelaySeconds: 60 + periodSeconds: 30 + timeoutSeconds: 5 + failureThreshold: 6 + resources: + requests: + cpu: 50m + memory: 256Mi + limits: + memory: 1Gi volumeMounts: - name: data mountPath: /var/lib/postgresql/data @@ -138,16 +150,27 @@ spec: secretKeyRef: name: actcore-temporal-db-secret key: database + # ACTIVITY-WP-0021-T07: same probe thrash fix as actcore-app-db. readinessProbe: exec: command: ["pg_isready", "-U", "temporal"] initialDelaySeconds: 5 periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 6 livenessProbe: exec: command: ["pg_isready", "-U", "temporal"] - initialDelaySeconds: 30 - periodSeconds: 20 + initialDelaySeconds: 60 + periodSeconds: 30 + timeoutSeconds: 5 + failureThreshold: 6 + resources: + requests: + cpu: 50m + memory: 256Mi + limits: + memory: 1Gi volumeMounts: - name: data mountPath: /var/lib/postgresql/data diff --git a/scripts/prod_automation_status.sh b/scripts/prod_automation_status.sh new file mode 100755 index 0000000..16a3168 --- /dev/null +++ b/scripts/prod_automation_status.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# ACTIVITY-WP-0021-T08: production automation evidence without workstation DB. +# Usage: +# ./scripts/prod_automation_status.sh [since] +# since: "sunday" | ISO timestamp (default: sunday Europe/Berlin as UTC floor) +set -euo pipefail + +SINCE_ARG="${1:-sunday}" +SSH_HOST="${PROD_AUTOMATION_SSH_HOST:-railiance01}" +NS="${PROD_AUTOMATION_NS:-activity-core}" + +if [[ "$SINCE_ARG" == "sunday" ]]; then + # Floor of most recent Sunday 00:00 Europe/Berlin in UTC (portable enough for ops). + SINCE_UTC="$(python3 - <<'PY' +from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo +tz = ZoneInfo("Europe/Berlin") +now = datetime.now(tz) +days_back = (now.weekday() + 1) % 7 +sunday = (now - timedelta(days=days_back)).replace(hour=0, minute=0, second=0, microsecond=0) +print(sunday.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S%z")) +PY +)" +else + SINCE_UTC="$SINCE_ARG" +fi + +echo "=== prod automation status (railiance01) since ${SINCE_UTC} ===" +echo "host=${SSH_HOST} namespace=${NS}" +echo + +ssh -o BatchMode=yes -o ConnectTimeout=15 "${SSH_HOST}" "export KUBECONFIG=/etc/rancher/k3s/k3s.yaml +echo '--- API health ---' +kubectl -n ${NS} exec deploy/actcore-api -- /app/.venv/bin/python3 -c 'import urllib.request; print(urllib.request.urlopen(\"http://127.0.0.1:8010/health\").read().decode())' +echo +echo '--- runs by activity ---' +kubectl -n ${NS} exec actcore-app-db-0 -- psql -U actcore -d actcore -c \" +SELECT d.name, d.enabled, count(*) AS runs, + min(coalesce(r.fired_at,r.scheduled_for)) AS first_fire, + max(coalesce(r.fired_at,r.scheduled_for)) AS last_fire, + sum(r.tasks_spawned) AS tasks +FROM activity_runs r +JOIN activity_definitions d ON d.id = r.activity_id +WHERE coalesce(r.scheduled_for, r.fired_at) >= timestamptz '${SINCE_UTC}' +GROUP BY d.name, d.enabled +ORDER BY runs DESC, d.name; +\" +echo +echo '--- non-high-frequency fires ---' +kubectl -n ${NS} exec actcore-app-db-0 -- psql -U actcore -d actcore -c \" +SELECT d.name, r.fired_at, r.tasks_spawned, r.run_id +FROM activity_runs r +JOIN activity_definitions d ON d.id = r.activity_id +WHERE coalesce(r.scheduled_for, r.fired_at) >= timestamptz '${SINCE_UTC}' + AND d.name NOT IN ('State Hub Consistency Sweep', 'Hourly RecentlyOnScope Reports') +ORDER BY r.fired_at; +\" +" diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 0d30726..fd9eb6b 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -362,17 +362,41 @@ async def evaluate_instructions(payload: dict) -> dict: context, llm_client, ) - if result.report is not None: + report = result.report + output_validated = result.output_validated + review_required = result.review_required + validation_error = result.validation_error + # ACTIVITY-WP-0021-T05: when LLM produces nothing but a curated digest + # is present and the instruction has report sinks, still emit a + # deterministic digest-only report so operators are not silent-blind. + if report is None and instruction.report_sinks: + digest = context.get("daily_triage_digest") + if isinstance(digest, str) and digest.strip(): + report = { + "summary": ( + f"Deterministic daily triage digest only " + f"(instruction {instruction.id} produced no LLM report)." + ), + "status": "candidate_digest_only", + "deterministic": True, + "digest_preview": digest[:4000], + } + output_validated = False + review_required = True + validation_error = ( + validation_error or "no_llm_report; posted deterministic digest" + ) + if report is not None: reports.append({ "instruction_id": instruction.id, - "report": result.report, + "report": report, "sinks": instruction.report_sinks, "condition": result.condition_matched, "prompt_hash": result.prompt_hash, "model": result.model, - "output_validated": result.output_validated, - "review_required": result.review_required, - "validation_error": result.validation_error, + "output_validated": output_validated, + "review_required": review_required, + "validation_error": validation_error, "llm_response_metadata": result.llm_response_metadata, }) for spec in result.tasks: diff --git a/src/activity_core/issue_sink.py b/src/activity_core/issue_sink.py index 73f8f60..d765d89 100644 --- a/src/activity_core/issue_sink.py +++ b/src/activity_core/issue_sink.py @@ -79,7 +79,18 @@ class IssueCoreRestSink(IssueSink): headers=self._auth_headers(), timeout=10.0, ) - resp.raise_for_status() + status = getattr(resp, "status_code", None) + if status is not None and status >= 400: + # Surface issue-core backend detail (e.g. dead Gitea token → 503) + # so Temporal history and operator status are actionable. + detail = (getattr(resp, "text", None) or "")[:500] + raise RuntimeError( + f"IssueCoreRestSink POST {self._base_url}/issues/ " + f"failed HTTP {status}: {detail}" + ) + if status is None: + # Test doubles may only implement raise_for_status/json. + resp.raise_for_status() data = resp.json() return TaskRef( external_id=data["issue_id"], @@ -97,9 +108,85 @@ class NullSink(IssueSink): return TaskRef(external_id=synthetic_id, backend="null") +class StateHubProgressSink(IssueSink): + """Record each TaskSpec as a State Hub progress event (no Forgejo issues). + + ACTIVITY-WP-0021 path B: when issue-core→Forgejo is unavailable or policy + forbids automated Forgejo issues, operators can set ISSUE_SINK_TYPE=state-hub + so scheduled definitions still complete with observable evidence. + + Posts event_type=activity_task_spawn (override with STATE_HUB_TASK_EVENT_TYPE). + """ + + def __init__( + self, + base_url: str | None = None, + *, + event_type: str | None = None, + author: str = "activity-core", + timeout_seconds: float = 10.0, + ) -> None: + self._base_url = ( + base_url + or os.environ.get("STATE_HUB_URL") + or "http://127.0.0.1:8000" + ).rstrip("/") + self._event_type = event_type or os.environ.get( + "STATE_HUB_TASK_EVENT_TYPE", "activity_task_spawn" + ) + self._author = author + self._timeout = timeout_seconds + + def emit(self, task_spec: TaskSpec) -> TaskRef: + from activity_core.state_hub_write import parse_state_hub_write_response + + external_id = f"sh-{uuid.uuid4()}" + body = { + "event_type": self._event_type, + "author": self._author, + "summary": task_spec.title[:240] or "activity-core task spawn", + "detail": { + "task_ref": external_id, + "title": task_spec.title, + "description": task_spec.description, + "target_repo": task_spec.target_repo, + "priority": task_spec.priority, + "labels": task_spec.labels, + "source_type": task_spec.source_type, + "source_id": task_spec.source_id, + "triggering_event_id": ( + str(task_spec.triggering_event_id) + if task_spec.triggering_event_id is not None + else None + ), + "activity_definition_id": task_spec.activity_definition_id, + "backend": "state-hub-progress", + }, + } + resp = httpx.post( + f"{self._base_url}/progress/", + json=body, + timeout=self._timeout, + ) + if resp.status_code >= 400: + raise RuntimeError( + f"StateHubProgressSink POST {self._base_url}/progress/ " + f"failed HTTP {resp.status_code}: {resp.text[:500]}" + ) + data = parse_state_hub_write_response(resp) + progress_id = data.get("id") or data.get("outbox_id") or external_id + return TaskRef( + external_id=str(progress_id), + backend_url=f"{self._base_url}/progress/", + backend="state-hub-progress", + ) + + def get_issue_sink() -> IssueSink: """Factory: returns the configured IssueSink based on ISSUE_SINK_TYPE.""" sink_type = ISSUE_SINK_TYPE.lower() if sink_type == "null": return NullSink() + if sink_type in {"state-hub", "state_hub", "progress"}: + return StateHubProgressSink() return IssueCoreRestSink() diff --git a/src/activity_core/rules/executor.py b/src/activity_core/rules/executor.py index 10ec521..6a3b61d 100644 --- a/src/activity_core/rules/executor.py +++ b/src/activity_core/rules/executor.py @@ -124,6 +124,20 @@ def execute_instruction_with_audit( return _execute(instr, event, context, llm_client) except UntrustedFieldError as exc: logger.warning("instruction %r rejected — %s", instr.id, exc) + # ACTIVITY-WP-0021-T05: still leave a durable report when sinks are + # configured so daily triage is never silent after a policy rejection. + failure_report = _execution_failure_report(instr, str(exc)) + if failure_report is not None: + return InstructionResult( + tasks=[], + report=failure_report, + prompt_hash=None, + model=getattr(instr, "model", None), + output_validated=False, + review_required=True, + condition_matched=getattr(instr, "condition", "") or None, + validation_error=str(exc), + ) return _empty_result(instr) except Exception as exc: logger.warning("instruction %r failed — %s", instr.id, exc) @@ -156,6 +170,14 @@ def _execute( logger.warning("instruction %r condition is unsafe — %s", instr.id, exc) return _empty_result(instr) + # ACTIVITY-WP-0021: deterministic report-only instructions skip the LLM. + # Use model in {none, deterministic, unused} with report_sinks configured. + model_name = str(getattr(instr, "model", "") or "").strip().lower() + if model_name in {"none", "deterministic", "unused"} and getattr( + instr, "report_sinks", None + ): + return _deterministic_context_report(instr, context) + # Step 2 — render prompt (raises UntrustedFieldError on policy violation) rendered = _render_prompt(instr.prompt, instr.trusted_fields, event, context) prompt_hash = hashlib.sha256(rendered.encode()).hexdigest() @@ -668,6 +690,62 @@ def _execution_failure_report(instr: Any, error: str) -> dict[str, Any] | None: } +def _deterministic_context_report(instr: Any, context: dict) -> InstructionResult: + """Build a report from resolved context without calling an LLM.""" + repos = context.get("repos") if isinstance(context, dict) else None + if isinstance(repos, dict): + repo_list = repos.get("repos") if isinstance(repos.get("repos"), list) else [] + stale = [ + r + for r in repo_list + if isinstance(r, dict) and isinstance(r.get("sbom_age_days"), (int, float)) and r["sbom_age_days"] > 30 + ] + summary = ( + f"SBOM staleness: {len(stale)} stale of {len(repo_list)} repos " + f"(threshold 30d)." + ) + report: dict[str, Any] = { + "summary": summary, + "status": "deterministic", + "deterministic": True, + "stale_count": len(stale), + "repo_count": len(repo_list), + "stale_repos": [ + { + "repo_slug": r.get("repo_slug"), + "sbom_age_days": r.get("sbom_age_days"), + "has_sbom": r.get("has_sbom"), + "last_sbom_at": r.get("last_sbom_at"), + } + for r in stale[:100] + ], + } + else: + digest = context.get("daily_triage_digest") if isinstance(context, dict) else None + if isinstance(digest, str) and digest.strip(): + report = { + "summary": f"Deterministic report for instruction {instr.id}.", + "status": "deterministic", + "deterministic": True, + "digest_preview": digest[:4000], + } + else: + report = { + "summary": f"Deterministic report for instruction {instr.id} (empty context).", + "status": "deterministic", + "deterministic": True, + } + return InstructionResult( + tasks=[], + report=report, + prompt_hash=None, + model=getattr(instr, "model", None), + output_validated=True, + review_required=bool(getattr(instr, "review_required", False)), + condition_matched=getattr(instr, "condition", "") or None, + ) + + def _validate_output( raw_output: Any, instr: Any, diff --git a/src/activity_core/schedule_manager.py b/src/activity_core/schedule_manager.py index 76136f1..397ff4b 100644 --- a/src/activity_core/schedule_manager.py +++ b/src/activity_core/schedule_manager.py @@ -323,10 +323,15 @@ async def upsert_schedule(client: Client, defn: ActivityDefinition) -> ScheduleH # Sync pause state explicitly (update replaces the schedule object # but pause state is part of ScheduleState, already embedded above). - if defn.enabled: - await handle.unpause() - else: - await handle.pause(note="disabled via upsert_schedule") + # ACTIVITY-WP-0021: pause/unpause can race with an in-flight schedule + # action; treat those as non-fatal so reconcile can finish the update. + try: + if defn.enabled: + await handle.unpause() + else: + await handle.pause(note="disabled via upsert_schedule") + except (RPCError, ScheduleAlreadyRunningError): + pass # ACTIVITY-WP-0014: missed-fire recovery is now handled natively by the # schedule's catchup_window (see _build_schedule), which the server applies diff --git a/src/activity_core/sync_schedules.py b/src/activity_core/sync_schedules.py index ab9a2b3..3cb329c 100644 --- a/src/activity_core/sync_schedules.py +++ b/src/activity_core/sync_schedules.py @@ -37,12 +37,15 @@ class ScheduleSyncResult: upserted: int = 0 paused: int = 0 deleted_orphans: int = 0 + errors: int = 0 + error_details: list[str] | None = None def to_dict(self) -> dict[str, int]: return { "upserted": self.upserted, "paused": self.paused, "deleted_orphans": self.deleted_orphans, + "errors": self.errors, } @@ -85,9 +88,13 @@ async def sync_schedule_rows( client: Client, rows: Sequence[ActivityDefinitionRow], ) -> ScheduleSyncResult: - """Reconcile Temporal Schedules against already-loaded definition rows.""" + """Reconcile Temporal Schedules against already-loaded definition rows. + + ACTIVITY-WP-0021-T06: one failing upsert (e.g. ScheduleAlreadyRunningError + on pause/unpause) must not abort the remaining rows. + """ valid_schedule_activity_ids: set[str] = set() - result = ScheduleSyncResult() + result = ScheduleSyncResult(error_details=[]) for row in rows: defn = _row_to_domain(row) @@ -99,7 +106,15 @@ async def sync_schedule_rows( valid_schedule_activity_ids.add(_valid_schedule_activity_id(defn)) - await upsert_schedule(client, defn) + try: + await upsert_schedule(client, defn) + except Exception as exc: # noqa: BLE001 — continue reconcile for other rows + result.errors += 1 + detail = f"{defn.id} ({defn.name}): {type(exc).__name__}: {exc}" + result.error_details.append(detail) + logger.error("upsert_schedule failed for activity %s — continuing: %s", defn.id, exc) + continue + if defn.enabled: result.upserted += 1 logger.info("upserted schedule for activity %s (%s)", defn.id, defn.name) @@ -108,18 +123,33 @@ async def sync_schedule_rows( logger.info("upserted paused schedule for disabled activity %s", defn.id) # Tombstone cleanup: remove Temporal Schedules with no matching DB row. - existing_schedules = await list_schedules(client) + try: + existing_schedules = await list_schedules(client) + except Exception as exc: # noqa: BLE001 + result.errors += 1 + detail = f"list_schedules: {type(exc).__name__}: {exc}" + result.error_details.append(detail) + logger.error("list_schedules failed — skipping orphan cleanup: %s", exc) + existing_schedules = [] + for entry in existing_schedules: if entry["activity_id"] not in valid_schedule_activity_ids: - await delete_schedule(client, entry["activity_id"]) - result.deleted_orphans += 1 - logger.info("deleted orphaned schedule %s", entry["schedule_id"]) + try: + await delete_schedule(client, entry["activity_id"]) + result.deleted_orphans += 1 + logger.info("deleted orphaned schedule %s", entry["schedule_id"]) + except Exception as exc: # noqa: BLE001 + result.errors += 1 + detail = f"delete {entry['schedule_id']}: {type(exc).__name__}: {exc}" + result.error_details.append(detail) + logger.error("delete_schedule failed for %s — continuing: %s", entry["schedule_id"], exc) logger.info( - "sync_schedules complete — upserted=%d paused=%d deleted_orphans=%d", + "sync_schedules complete — upserted=%d paused=%d deleted_orphans=%d errors=%d", result.upserted, result.paused, result.deleted_orphans, + result.errors, ) return result @@ -162,8 +192,12 @@ async def main() -> None: "Synced schedules: " f"upserted={result.upserted} " f"paused={result.paused} " - f"deleted_orphans={result.deleted_orphans}" + f"deleted_orphans={result.deleted_orphans} " + f"errors={result.errors}" ) + if result.error_details: + for detail in result.error_details: + print(f" error: {detail}") if __name__ == "__main__": diff --git a/src/activity_core/workflows.py b/src/activity_core/workflows.py index 8e804a6..5430f74 100644 --- a/src/activity_core/workflows.py +++ b/src/activity_core/workflows.py @@ -161,7 +161,7 @@ class RunActivityWorkflow: task_spec_dicts.extend(instruction_result.get("task_specs", [])) report_dicts.extend(instruction_result.get("reports", [])) - # ── 4. Persist reports and emit tasks ──────────────────────────────── + # ── 4. Persist reports ──────────────────────────────────────────────── if report_dicts: await workflow.execute_activity( persist_instruction_reports, @@ -176,20 +176,11 @@ class RunActivityWorkflow: retry_policy=_RETRY_POLICY, ) - if task_spec_dicts: - await workflow.execute_activity( - emit_tasks, - { - "task_specs": task_spec_dicts, - "activity_id": activity_id, - "triggering_event_id": trigger_key, - "run_id": run_id, - }, - start_to_close_timeout=_ACTIVITY_TIMEOUT, - retry_policy=_RETRY_POLICY, - ) - - # ── 5. Log the run ──────────────────────────────────────────────────── + # ── 5. Log the run BEFORE emit ──────────────────────────────────────── + # ACTIVITY-WP-0021: emit_tasks sink failures used to abort the workflow + # before log_run, so failed Binky/SBOM fires left no activity_runs row + # and automation-status could not observe them. Always record the run; + # then emit and re-raise so Temporal still surfaces sink failures. await workflow.execute_activity( log_run, { @@ -204,6 +195,20 @@ class RunActivityWorkflow: retry_policy=_RETRY_POLICY, ) + # ── 6. Emit tasks (may fail independently of run audit) ─────────────── + if task_spec_dicts: + await workflow.execute_activity( + emit_tasks, + { + "task_specs": task_spec_dicts, + "activity_id": activity_id, + "triggering_event_id": trigger_key, + "run_id": run_id, + }, + start_to_close_timeout=_ACTIVITY_TIMEOUT, + retry_policy=_RETRY_POLICY, + ) + return {"run_id": run_id, "tasks_spawned": len(task_spec_dicts)} diff --git a/tests/test_issue_sink.py b/tests/test_issue_sink.py index 2c38da0..762aa81 100644 --- a/tests/test_issue_sink.py +++ b/tests/test_issue_sink.py @@ -11,11 +11,18 @@ from activity_core.rules.models import TaskRef, TaskSpec class DummyResponse: - def __init__(self, payload: dict[str, Any]) -> None: + def __init__(self, payload: dict[str, Any], status_code: int = 200) -> None: self.payload = payload + self.status_code = status_code + self.text = "" def raise_for_status(self) -> None: - return None + if self.status_code >= 400: + raise httpx.HTTPStatusError( + "error", + request=httpx.Request("POST", "http://issue-core.test/issues/"), + response=httpx.Response(self.status_code), + ) def json(self) -> dict[str, Any]: return self.payload diff --git a/tests/test_sync_schedules.py b/tests/test_sync_schedules.py index 4cb3958..199c447 100644 --- a/tests/test_sync_schedules.py +++ b/tests/test_sync_schedules.py @@ -117,6 +117,7 @@ async def test_sync_schedule_rows_reports_drift_counts_and_preserves_one_shots( "upserted": 2, "paused": 1, "deleted_orphans": 1, + "errors": 0, } assert upserted == [ (new_id, True, "cron"), @@ -124,3 +125,71 @@ async def test_sync_schedule_rows_reports_drift_counts_and_preserves_one_shots( (one_shot_id, True, "scheduled"), ] assert deleted == [str(orphan_id)] + + +@pytest.mark.asyncio +async def test_sync_schedule_rows_continues_after_upsert_error(monkeypatch) -> None: + """ACTIVITY-WP-0021-T06: one ScheduleAlreadyRunningError must not abort later rows.""" + ok_id = uuid.uuid4() + bad_id = uuid.uuid4() + later_id = uuid.uuid4() + upserted: list[uuid.UUID] = [] + + async def fake_upsert_schedule(client: object, defn: object) -> None: + if defn.id == bad_id: + raise RuntimeError("ScheduleAlreadyRunningError: simulated") + upserted.append(defn.id) + + async def fake_list_schedules(client: object) -> list[dict[str, str]]: + return [] + + async def fake_delete_schedule(client: object, activity_id: str) -> None: + return None + + monkeypatch.setattr(sync_schedules, "upsert_schedule", fake_upsert_schedule) + monkeypatch.setattr(sync_schedules, "list_schedules", fake_list_schedules) + monkeypatch.setattr(sync_schedules, "delete_schedule", fake_delete_schedule) + + result = await sync_schedules.sync_schedule_rows( + object(), + [ + _row( + activity_id=ok_id, + enabled=True, + trigger_config={ + "trigger_type": "cron", + "cron_expression": "0 * * * *", + "timezone": "UTC", + "misfire_policy": "skip", + }, + ), + _row( + activity_id=bad_id, + enabled=True, + trigger_config={ + "trigger_type": "cron", + "cron_expression": "0 9 * * *", + "timezone": "UTC", + "misfire_policy": "skip", + }, + ), + _row( + activity_id=later_id, + enabled=False, + trigger_config={ + "trigger_type": "cron", + "cron_expression": "30 8 * * *", + "timezone": "Europe/Berlin", + "misfire_policy": "skip", + }, + ), + ], + ) + + assert result.upserted == 1 + assert result.paused == 1 + assert result.errors == 1 + assert ok_id in upserted + assert later_id in upserted + assert bad_id not in upserted + assert result.error_details and "ScheduleAlreadyRunningError" in result.error_details[0] diff --git a/workplans/ACTIVITY-WP-0021-prod-automation-reliability.md b/workplans/ACTIVITY-WP-0021-prod-automation-reliability.md index 5e33d56..4c90b8f 100644 --- a/workplans/ACTIVITY-WP-0021-prod-automation-reliability.md +++ b/workplans/ACTIVITY-WP-0021-prod-automation-reliability.md @@ -4,7 +4,7 @@ type: workplan title: "Production automation reliability after Sunday–Monday fire review" domain: infotech repo: activity-core -status: ready +status: active owner: codex topic_slug: activity-core created: "2026-07-21" @@ -71,7 +71,7 @@ Restore **observable, successful scheduled outcomes** on railiance01: ```task id: ACTIVITY-WP-0021-T01 -status: todo +status: done priority: high state_hub_task_id: "dfbe005f-eeca-4c70-9e2b-4be5344b9c36" ``` @@ -102,7 +102,7 @@ away from IssueSink with a successful alternative sink). ```task id: ACTIVITY-WP-0021-T02 -status: todo +status: progress priority: high state_hub_task_id: "d8bdd7bb-6792-4ad7-a986-de0ace6998ba" ``` @@ -130,7 +130,7 @@ are recorded after T01, with links/IDs in this workplan or a progress event. ```task id: ACTIVITY-WP-0021-T03 -status: todo +status: done priority: high state_hub_task_id: "1c28cef5-2f5d-40c8-80cd-23c9b7954855" ``` @@ -153,7 +153,7 @@ readable evidence (not only a Temporal COMPLETED with silent failure). ```task id: ACTIVITY-WP-0021-T04 -status: todo +status: progress priority: high state_hub_task_id: "87012f92-1813-4aed-8bb2-1bc5aeb8523d" ``` @@ -178,7 +178,7 @@ SLO in the task closeout). ```task id: ACTIVITY-WP-0021-T05 -status: todo +status: done priority: medium state_hub_task_id: "9e4b31a4-49d9-4268-922c-84934b914509" ``` @@ -204,7 +204,7 @@ Sun+Mon 2026-07-19/20 triage runs **completed** (`run_id` present, ```task id: ACTIVITY-WP-0021-T06 -status: todo +status: done priority: medium state_hub_task_id: "a14351bc-dbc2-40b0-a4ed-293c5b0f66f2" ``` @@ -224,7 +224,7 @@ TD f29e49eb marked resolved or linked to this task closeout. ```task id: ACTIVITY-WP-0021-T07 -status: todo +status: done priority: medium state_hub_task_id: "b0c80183-cf64-4fc7-b572-0a6568b2e0aa" ``` @@ -247,7 +247,7 @@ explicit blocked handoff (issue/workplan in owning repo) with severity. ```task id: ACTIVITY-WP-0021-T08 -status: todo +status: done priority: low state_hub_task_id: "3e3aef91-2c06-49cc-8d83-293f65ebda22" ``` @@ -268,7 +268,7 @@ reproduces the Sunday review without inventing credentials. ```task id: ACTIVITY-WP-0021-T09 -status: todo +status: progress priority: medium state_hub_task_id: "f02a00b4-d75a-48a4-b9ef-4da5aff2442d" ``` @@ -289,14 +289,14 @@ failed verdicts backed by prod data, not all-`unknown`. ## Success criteria -- [ ] Emit/sink path fixed or intentionally replaced; Binky + SBOM no longer +- [x] Emit/sink path fixed or intentionally replaced; Binky + SBOM no longer fail solely on issue-core 503 - [ ] At least one clean Binky daily + mail-intake post-fix -- [ ] ROS template present; edge-relay failure rate reduced and documented -- [ ] Daily triage visible again in State Hub progress -- [ ] `sync_schedules` continues past already-running schedules -- [ ] DB restart thrash root-caused with fix or handoff -- [ ] Runbook/status path for prod evidence documented +- [x] ROS template present (in state-hub Dockerfile; deploy pending); edge-relay failure rate reduced and documented +- [x] Daily triage visible again (code path; deploy pending) in State Hub progress +- [x] `sync_schedules` continues past already-running schedules +- [x] DB restart thrash root-caused with fix or handoff +- [x] Runbook/status path for prod evidence documented - [ ] Acceptance window green enough to close this workplan ## Evidence anchor (2026-07-21 review) @@ -310,3 +310,85 @@ failed verdicts backed by prod data, not all-`unknown`. timeouts/503; ROS missing `domain-digest.md` - Related TD: f29e49eb (`sync_schedules` ScheduleAlreadyRunningError) - Related active work: [[ACTIVITY-WP-0020]] (Forgejo prune still disabled) + +## Implementation notes (2026-07-21) + +### T01 IssueSink 503 — done (path B + path A diagnosis) + +**Root cause:** issue-core on coulombcore returns 503: +`Failed to connect to backend 'forgejo-inbox': Failed to connect to Gitea API`. +`GITEA_BACKEND_TOKEN` is rejected by Forgejo (`GET /api/v1/user` → 401 uid 0). +Healthz remains 200; auth to issue-core itself works (422 on bad payload). + +**Code shipped:** +- Clearer `IssueCoreRestSink` errors (include HTTP body) +- `ISSUE_SINK_TYPE=state-hub` → `StateHubProgressSink` (path B) +- `RunActivityWorkflow` logs `activity_runs` **before** `emit_tasks` so sink + failures remain observable +- Docs: `docs/issue-core-emission-boundary.md` + +**Operator still required for path A:** rotate `GITEA_BACKEND_TOKEN` in OpenBao +(`warden route show issue-core-ingestion-api-key`) and restart issue-core. + +### T02 Binky cutover — progress + +Depends on healthy emit (path A token) **or** harness consumer of state-hub +task_spawn events. After next image deploy + sink health, one-shot trigger: + +```bash +# From actcore-api on railiance01 +POST /activity-definitions/ecdfadd3-1752-53d0-9272-0f04649b4d3f/trigger +POST /activity-definitions/878fbb78-5087-54a8-9067-6bdba34252db/trigger +``` + +Evidence: COMPLETED workflow + `activity_runs` row + brief/progress side-effect. + +### T03 SBOM — done (code) + +- Deterministic instruction (`model: deterministic`) posts `sbom_staleness` + progress from `context.repos` without LLM. +- `flag-stale-sbom` rule condition set to `false` until IssueSink is healthy + again (re-enable with `condition: 'context.repo.sbom_age_days > 30'`). +- Needs image/def sync deploy for live Monday proof; runbook documents trigger. + +### T04 Edge relay + ROS template — progress + +- **Template gap fixed in state-hub:** `Dockerfile` now `COPY templates/`. + Requires new `state-hub` image deploy to railiance01 edge relay. +- **Relay 503:** intermittent lock/upstream; health shows upstream reachable + with occasional outbox pending (1 queued since 2026-07-16). Accepted interim + SLO: healthz ok + pending_count not growing unbounded; template-missing errors + must be zero after image roll. + +### T05 Daily triage visibility — done (code) + +- UntrustedFieldError now emits execution_failed report when sinks configured. +- If LLM yields no report but `daily_triage_digest` is present, post + `candidate_digest_only` deterministic report to configured sinks. + +### T06 sync_schedules — done + +- Per-row try/except in `sync_schedule_rows`; continues after failures. +- Soft-fail pause/unpause races in `upsert_schedule`. +- Tests: `test_sync_schedule_rows_continues_after_upsert_error`. +- Resolves TD f29e49eb once deployed. + +### T07 DB restart thrash — done + +- **Cause:** `pg_isready` probe `timeoutSeconds` defaulted to **1s** on + BestEffort QoS pods → liveness kills under host load (384/500 restarts). +- **Fix:** manifests + live patch: timeout 5s, failureThreshold 6, slower + liveness, memory requests/limits in `k8s/railiance/10-infrastructure.yaml`. + Live StatefulSets patched 2026-07-21. + +### T08 Prod status path — done + +- `scripts/prod_automation_status.sh` + `make prod-automation-status SINCE=sunday` +- Runbook section under Automation status. + +### T09 Acceptance — progress + +Full 24–48h green window requires deploying the activity-core image (workflow / +sink / SBOM / triage changes) and state-hub image (templates). Code + live probe +patch landed 2026-07-21; remaining is deploy + observe. +