Implement ACTIVITY-WP-0022/0023: safe sink default and gap closures

Default ISSUE_SINK_TYPE to state-hub (no silent Forgejo issues), hard-fail
prune apply without live-images protection, refresh-live-images script,
disable TaskExecutor stub by default, and document consumer/sink contracts.

src/activity_core/context_resolvers/forgejo_prune.py

@@ -29,17 +29,30 @@ def forgejo_package_prune(params: dict[str, Any]) -> dict[str, Any]:
     if apply:
         cmd.append("--apply")
 
-    # ACTIVITY-WP-0020: worker pods often lack kubectl, so live-tag protection
-    # must come from an explicit multi-cluster image list file (hostPath).
+    # ACTIVITY-WP-0020 / 0023-T03: worker pods often lack kubectl, so live-tag
+    # protection must come from an explicit multi-cluster image list file.
+    # Apply is refused without a non-empty protection file (prevents the
+    # 2026-07-21 incident that deleted live state-hub tags).
     live_images = params.get("live_images_file") or os.environ.get(
         "FORGEJO_LIVE_IMAGES_FILE", ""
     )
+    live_path: Path | None = None
     if live_images:
         live_path = Path(str(live_images)).expanduser()
-        if live_path.is_file():
+        if live_path.is_file() and live_path.stat().st_size > 0:
             cmd.append(f"--live-images-file={live_path}")
+        elif apply:
+            raise RuntimeError(
+                f"forgejo_package_prune apply=true requires a non-empty "
+                f"live_images_file; missing or empty: {live_path}"
+            )
         else:
-            logger.warning("live_images_file not found: %s", live_path)
+            logger.warning("live_images_file not found or empty: %s", live_path)
+    elif apply:
+        raise RuntimeError(
+            "forgejo_package_prune apply=true requires params.live_images_file "
+            "or FORGEJO_LIVE_IMAGES_FILE (non-empty multi-cluster image list)"
+        )
 
     env = os.environ.copy()
     completed = subprocess.run(

src/activity_core/issue_sink.py

@@ -2,9 +2,17 @@
 IssueSink adapter interface and implementations.
 
 IssueSink is the outbound boundary between activity-core and task backends
-(issue-core, etc.). It receives TaskSpec objects and returns TaskRef objects.
+(issue-core, State Hub progress, etc.). It receives TaskSpec objects and
+returns TaskRef objects.
 
-Active sink is selected by ISSUE_SINK_TYPE env var: "rest" (default) | "null".
+Active sink is selected by ISSUE_SINK_TYPE:
+
+  state-hub (default) — State Hub progress (`activity_task_spawn`); no Forgejo
+  null                — dry-run synthetic refs
+  rest                — issue-core REST (explicit opt-in; may project to Forgejo)
+
+ACTIVITY-WP-0022: default must not silently create Forgejo issues for internal
+findings. Use rest only when intentionally projecting to an external tracker.
 """
 
 from __future__ import annotations
@@ -22,7 +30,9 @@ logger = logging.getLogger(__name__)
 
 ISSUE_CORE_URL = os.environ.get("ISSUE_CORE_URL", "http://127.0.0.1:8765")
 ISSUE_CORE_API_KEY_ENV = "ISSUE_CORE_API_KEY"
-ISSUE_SINK_TYPE = os.environ.get("ISSUE_SINK_TYPE", "rest")
+# Safe default for internal fleet findings (ACTIVITY-WP-0022).
+DEFAULT_ISSUE_SINK_TYPE = "state-hub"
+ISSUE_SINK_TYPE = os.environ.get("ISSUE_SINK_TYPE", DEFAULT_ISSUE_SINK_TYPE)
 
 
 class IssueSink(ABC):
@@ -188,9 +198,18 @@ def get_issue_sink() -> IssueSink:
     Re-reads the env on each call so ConfigMap/env patches apply without
     requiring a module reload (ACTIVITY-WP-0021).
     """
-    sink_type = os.environ.get("ISSUE_SINK_TYPE", ISSUE_SINK_TYPE).lower()
+    sink_type = os.environ.get("ISSUE_SINK_TYPE", DEFAULT_ISSUE_SINK_TYPE).strip().lower()
+    if not sink_type:
+        sink_type = DEFAULT_ISSUE_SINK_TYPE
     if sink_type == "null":
         return NullSink()
     if sink_type in {"state-hub", "state_hub", "progress"}:
         return StateHubProgressSink()
-    return IssueCoreRestSink()
+    if sink_type == "rest":
+        return IssueCoreRestSink()
+    logger.warning(
+        "unknown ISSUE_SINK_TYPE=%r — falling back to %s (safe default)",
+        sink_type,
+        DEFAULT_ISSUE_SINK_TYPE,
+    )
+    return StateHubProgressSink()

src/activity_core/worker.py

@@ -113,23 +113,51 @@ async def run() -> None:
         ],
     )
 
-    task_worker = Worker(
-        client,
-        task_queue=TASK_EXECUTION_TASK_QUEUE,
-        workflows=[TaskExecutorWorkflow],
-        activities=[persist_task_instance],
-    )
+    # ACTIVITY-WP-0023-T08: only register the legacy task-execution stub when
+    # explicitly enabled. Default is orchestrator-only so production does not
+    # advertise a fake execution surface on task-execution-tq.
+    enable_task_stub = os.environ.get(
+        "ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB", ""
+    ).strip().lower() in {"1", "true", "yes", "on"}
+
+    task_worker: Worker | None = None
+    if enable_task_stub:
+        task_worker = Worker(
+            client,
+            task_queue=TASK_EXECUTION_TASK_QUEUE,
+            workflows=[TaskExecutorWorkflow],
+            activities=[persist_task_instance],
+        )
+        logger.warning(
+            "TaskExecutorWorkflow stub ENABLED on %s — not for production execution",
+            TASK_EXECUTION_TASK_QUEUE,
+        )
+    else:
+        logger.info(
+            "TaskExecutorWorkflow stub not registered "
+            "(set ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB=true for legacy tests only)"
+        )
 
     loop = asyncio.get_running_loop()
     stop = asyncio.Event()
     loop.add_signal_handler(signal.SIGTERM, stop.set)
     loop.add_signal_handler(signal.SIGINT, stop.set)
 
-    async with orchestrator_worker, task_worker:
+    workers = [orchestrator_worker]
+    if task_worker is not None:
+        workers.append(task_worker)
+
+    from contextlib import AsyncExitStack
+
+    async with AsyncExitStack() as stack:
+        for w in workers:
+            await stack.enter_async_context(w)
+        queues = [ORCHESTRATOR_TASK_QUEUE]
+        if enable_task_stub:
+            queues.append(TASK_EXECUTION_TASK_QUEUE)
         logger.info(
-            "Workers running — queues: %r, %r (namespace=%r)",
-            ORCHESTRATOR_TASK_QUEUE,
-            TASK_EXECUTION_TASK_QUEUE,
+            "Workers running — queues: %r (namespace=%r)",
+            queues,
             TEMPORAL_NAMESPACE,
         )
         await stop.wait()

src/activity_core/workflows.py

@@ -214,26 +214,45 @@ class RunActivityWorkflow:
 
 @workflow.defn
 class TaskExecutorWorkflow:
-    """Compatibility stub for legacy task-instance workflows.
+    """LEGACY NO-OP — not a production execution surface (ACTIVITY-WP-0023-T08).
 
-    This is not a production execution surface for activity-core. It persists a
-    task_instances row with status=done and returns immediately so legacy/dev
-    flows keep their idempotency behavior. Real task execution belongs in
-    per-repo workers or a future execution-owned repo/workplan, not here.
+    Historical compatibility stub. Real task execution belongs in per-repo
+    workers / agent-harness, not activity-core.
 
-    task_id is derived deterministically from the workflow's own ID so
-    persist_task_instance retries remain idempotent.
+    Behaviour is controlled by ``ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB``:
+
+    - unset / false (default): refuse to run (logs error, raises) so the stub
+      cannot attract production work by accident.
+    - true: legacy behaviour — persist a done ``task_instances`` row for
+      idempotent dev/test callers only.
     """
 
     @workflow.run
     async def run(self, run_id: str, task_type: str, params: dict) -> dict:
-        # Keep the stub idempotent without implying task lifecycle ownership.
+        enabled = (
+            os.environ.get("ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB", "")
+            .strip()
+            .lower()
+            in {"1", "true", "yes", "on"}
+        )
         task_id = str(
             uuid.uuid5(uuid.NAMESPACE_URL, workflow.info().workflow_id)
         )
 
-        workflow.logger.info(
-            "TaskExecutorWorkflow started",
+        if not enabled:
+            workflow.logger.error(
+                "TaskExecutorWorkflow refused: stub disabled "
+                "(set ACTIVITY_CORE_ENABLE_TASK_EXECUTOR_STUB=true only for legacy tests). "
+                "Real execution belongs in per-repo workers / agent-harness. "
+                "See docs/task-emission-consumer-contract.md"
+            )
+            raise RuntimeError(
+                "TaskExecutorWorkflow is disabled (ACTIVITY-WP-0023-T08). "
+                "Use per-repo executors; do not route production work here."
+            )
+
+        workflow.logger.warning(
+            "TaskExecutorWorkflow stub running (legacy mode)",
             extra={"run_id": run_id, "task_type": task_type, "task_id": task_id},
         )
 
@@ -251,4 +270,4 @@ class TaskExecutorWorkflow:
             retry_policy=_RETRY_POLICY,
         )
 
-        return {"task_id": task_id, "status": "done"}
+        return {"task_id": task_id, "status": "done", "legacy_stub": True}