Track fenced Engine stages for the factory floor

docs/architecture.md

@@ -51,6 +51,16 @@ Source: [`api.py`](https://github.com/fireflyframework/firefly-forge/blob/main/s
 
 `Store.event()` retains the sanitized message separately from that bounded projection. Studio renders supported summaries, leaves plain logs readable and offers an explicit verified reader for complete content. Unsupported structured previews receive a generic card. The optional Claude summary fields preserve the existing descriptor version and byte/hash binding; the archive writer, current-authority checks, credential fences and D04 export limits are shared. See [the activity contract](agent-integrations.md#read-complete-retained-provider-activity) and [the operator walkthrough](run-evidence.md#read-complete-provider-activity).
 
+### Current Engine stage observations
+
+`EngineActivity` is a native pyfly service with a separate PostgreSQL observation table. After a new native task intent is admitted, Engine records the exact run, task, attempt, current plan and frozen scope. One heartbeat renews its current local tasks together. Receipts and shutdown end those observations. SQL checks the current canonical execution identity and project permission, the owning PostgreSQL backend and advisory fence, and the observation's expiry before the dashboard counts it as active.
+
+This describes an Engine-owned stage, including a stage awaiting another system. It does not establish remote runner CPU work, provider availability or a person's presence. The existing execution snapshot retains a canonical user ID and its durable grants; it does not retain the original browser cookie. Signing out revokes that browser session, not the queued job. Suspending the execution identity, removing its required project permission or losing its current administrator grant invalidates the corresponding observation. Dashboard readers independently need current read authority and, for cookie access, their exact active session after the snapshot read.
+
+Unsupported SCM-mapped and personal-token execution actors stay unknown. Personal read tokens can still observe their selected projects through the existing read boundary. Observation failure expires to unknown and cannot retry productive work, approve an effect or skip credential and ownership cleanup. The two new observation modules enter the prospective runtime/admission material; their presence does not register a coding profile.
+
+Source: `engine_activity.py`, `engine_activity_persistence.py`, `engine.py` (`_observe_stage`, `_begin_task`, `_receipt`) and `factory_dashboard_persistence.py`. See [the operator guide](factory-dashboard.md#engine-stage-observations) for freshness settings and the limits of the display.
+
 ### Durable natural ticket conversations
 
 `SCMNaturalTasks` retains a human message before its configured `communication`

docs/factory-dashboard.md

@@ -41,10 +41,26 @@ The stage inspector describes **tasks** at the selected station. **Runs in this
 
 **Recorded running** means the stored run state describes work in progress. It does not establish that a worker is currently alive. A task last recorded as running can have unknown current activity while its run still has a running state.
 
-The present projection cannot observe a fresh per-run worker lease. Execution-request claims are retained custody records; a factory-wide advisory lock is not a heartbeat for every task. The runner controller also owns a separate private ledger. The dashboard does not query that ledger by assuming a schema or adding cross-schema privileges. Machinery stays still when current activity cannot be established. Aggregate scopes never show a single package as if it represented every run.
+The dashboard can use a separate, current observation of an Engine-owned stage. Execution-request claims alone are retained custody records; a factory-wide advisory lock alone is not a heartbeat for every task. Remote runners own a separate private ledger, which the dashboard does not query by assuming a schema or adding cross-schema privileges. Machinery stays still when current stage activity cannot be established. Aggregate scopes never show a single package as if it represented every run.
 
 Worker liveness, queue-wait duration, deployment health and runtime metrics remain unavailable in this view. Recorded expired or quarantined run states are separate from unknown states; they are not inferred runner-lease diagnoses. Deployment and support evidence have their own views and recorded sources.
 
+## Engine stage observations
+
+An observed active stage means the factory's Engine still owns that task and its observation has not expired. The stage may be waiting on a runner or model; the display does not establish that the remote system is using CPU. A missing, expired or inconsistent observation stays unknown. Grouped activity expires when its earliest supporting observation expires.
+
+Signing out does not cancel a queued or running job. Execution uses its retained canonical identity and current grants. Suspend that identity or remove its required project permission to withdraw that authority; use the run's cancellation action when cancellation is intended. The person reading this page still needs current project access and an active browser session. SCM-mapped and personal-token execution actors are not treated as supported activity producers. A personal read token grants observation only.
+
+Administrators can configure these server settings before starting forge:
+
+| Setting | Default | Purpose |
+| --- | --- | --- |
+| `FORGE_ACTIVITY_HEARTBEAT_SECONDS` | `5` | Interval between samples for locally owned Engine stages. |
+| `FORGE_ACTIVITY_STALE_SECONDS` | `20` | Freshness window; it must exceed the heartbeat interval. |
+| `FORGE_ACTIVITY_SAMPLE_TIMEOUT_SECONDS` | `1` | Time allowed for an observation sample before it becomes unavailable. |
+
+These intervals govern telemetry, not the time allowed to implement a product. An unavailable observation does not stop or replay productive work. Worker health, external execution and final command outcomes retain their own evidence boundaries.
+
 ## Inspect a run
 
 1. Choose a product project, or keep the factory overview.

src/firefly_forge/agentic_checkpoint_invocation.py

@@ -249,6 +249,8 @@ def runtime_contract() -> str:
                     "identity_persistence.py",
                     "access_persistence.py",
                     "product_provenance.py",
+                    "engine_activity.py",
+                    "engine_activity_persistence.py",
                 )
             },
         }

src/firefly_forge/api.py

@@ -101,6 +101,8 @@
     SCMRepositorySelection,
 )
 from firefly_forge.engine import Engine
+from firefly_forge.engine_activity import EngineActivity, EngineActivitySettings
+from firefly_forge.engine_activity_persistence import ForgeEngineActivityRepository
 from firefly_forge.execution_request_persistence import (
     ForgeExecutionRequestEventRepository,
     ForgeExecutionRequestRepository,
@@ -779,6 +781,18 @@ def user_administration(
     ) -> UserAdministration:
         return UserAdministration(store, identity, operations, events)
 
+    @bean
+    def engine_activity(self, store: Store, records: ForgeEngineActivityRepository) -> EngineActivity:
+        return EngineActivity(
+            store,
+            records,
+            EngineActivitySettings(
+                heartbeat_seconds=float(os.environ.get("FORGE_ACTIVITY_HEARTBEAT_SECONDS", "5")),
+                stale_seconds=float(os.environ.get("FORGE_ACTIVITY_STALE_SECONDS", "20")),
+                sample_timeout_seconds=float(os.environ.get("FORGE_ACTIVITY_SAMPLE_TIMEOUT_SECONDS", "1")),
+            ),
+        )
+
     @bean
     def coordinator(
         self,
@@ -797,13 +811,15 @@ def coordinator(
         routing: TaskRouting,
         runtime_roles: RuntimeRoles,
         worker_evidence: WorkerEvidence,
+        stage_activity: EngineActivity,
     ) -> Engine:
         engine = Engine(
             store,
             self.data_dir / "workspaces",
             concurrency=int(os.environ.get("FORGE_CONCURRENCY", "2")),
             connections=connections,
             journal=journal,
+            stage_activity=stage_activity,
         )
         engine.assurance = assurance
         engine.worker_evidence = worker_evidence
@@ -1375,6 +1391,7 @@ class Application:
         ForgePersonalTokenEventRepository,
         PersonalTokenController,
         FactoryDashboardRepository,
+        ForgeEngineActivityRepository,
         FactoryDashboardController,
         ForgeMembershipRepository,
         ForgeAccessEventRepository,

src/firefly_forge/engine.py

@@ -6,13 +6,15 @@
 import asyncio
 import hashlib
 import json
+import logging
 import os
 import re
+import sys
 import uuid
 from collections.abc import Callable
 from contextlib import AsyncExitStack
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 import httpx
 from filelock import FileLock, Timeout
@@ -31,6 +33,7 @@
     profile_observation,
     trace_observation,
 )
+from firefly_forge.engine_activity import EngineActivity
 from firefly_forge.git_diff_models import GitDiffBudget
 from firefly_forge.git_workspace import git_environment, metadata_digest, trusted_git
 from firefly_forge.models import Command, NewWorkCommand, configured_commands
@@ -143,12 +146,14 @@ def __init__(
         concurrency: int = 2,
         connections: Any = None,
         journal: Any = None,
+        stage_activity: EngineActivity | None = None,
     ) -> None:
         if not 1 <= concurrency <= 64:
             raise ValueError("Execution concurrency must be between 1 and 64")
         self.store = store
         self.connections = connections
         self.journal = journal
+        self.stage_activity = stage_activity
         from firefly_forge.communication import CommunicationWriter
 
         self.communication_writer = CommunicationWriter(store)
@@ -280,12 +285,30 @@ async def stop(self) -> None:
             self.ownership_failure = "Execution consumers have not drained; storage ownership remains fenced"
             self.execution_state = "degraded"
             return
-        if self.enforce_ownership:
-            try:
-                await self.store.release_ownership()
-            finally:
-                self.owner_lock.release()
-                RETAINED_EXECUTION_FENCES.discard(self.owner_lock)
+        try:
+            await self._observe_stage("close")
+        finally:
+            if self.enforce_ownership:
+                try:
+                    await self.store.release_ownership()
+                finally:
+                    self.owner_lock.release()
+                    RETAINED_EXECUTION_FENCES.discard(self.owner_lock)
+
+    async def _observe_stage(self, operation: Literal["begin", "end", "close"], *args: str) -> None:
+        if self.stage_activity is None:
+            return
+        pending = sys.exception()
+        try:
+            await getattr(self.stage_activity, operation)(*args)
+        except asyncio.CancelledError:
+            task = asyncio.current_task()
+            # A cancelled observer is not a cancellation request for productive work.
+            if task is not None and task.cancelling() and not isinstance(pending, asyncio.CancelledError):
+                raise
+            logging.getLogger(__name__).warning("Engine stage observer unavailable; activity remains unknown")
+        except Exception:
+            logging.getLogger(__name__).warning("Engine stage observer unavailable; activity remains unknown")
 
     async def assert_ownership(self) -> None:
         if self.store.runtime_roles:
@@ -1682,7 +1705,10 @@ async def launch_guard(
                 (await self.store.update(identifier, publication_error=safe(str(exc))))
                 await emit("publication", f"Publication failed: {exc}")
         finally:
-            await credential_scopes.aclose()
+            try:
+                await self._observe_stage("end", identifier)
+            finally:
+                await credential_scopes.aclose()
 
     async def _worker_feedback(
         self,
@@ -1865,6 +1891,8 @@ async def _begin_task(
             payload.update(reason="verification_repair", failed_checks=failed_checks)
         intent = await self.journal.intent(identifier, task_id, attempt=attempt, payload=payload)
         if intent["disposition"] != "needs_reconciliation":
+            if intent["disposition"] == "execute":
+                await self._observe_stage("begin", identifier, task_id, intent["idempotency_key"])
             return dict(intent)
         run = await self.store.get_run(identifier)
         if not run.get("resume_existing"):
@@ -1893,6 +1921,7 @@ async def _receipt(
                 await self.journal._receipt(identifier, key, evidence, checkpoint=checkpoint, terminal=terminal)
             else:
                 await self.journal.receipt(identifier, key, evidence, checkpoint=checkpoint)
+            await self._observe_stage("end", identifier, key)
 
     @staticmethod
     def component_directory(run: dict[str, Any], workspace: Path, *, create: bool = False) -> Path: