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.

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:

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()

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,

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

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__":

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)}