diff --git a/docs/architecture.md b/docs/architecture.md index 55d984dae..418d65628 100644 --- a/docs/architecture.md +++ b/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` diff --git a/docs/factory-dashboard.md b/docs/factory-dashboard.md index f86d3e68b..ef5bb5eec 100644 --- a/docs/factory-dashboard.md +++ b/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. diff --git a/src/firefly_forge/agentic_checkpoint_invocation.py b/src/firefly_forge/agentic_checkpoint_invocation.py index 264363f8c..8049ca841 100644 --- a/src/firefly_forge/agentic_checkpoint_invocation.py +++ b/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", ) }, } diff --git a/src/firefly_forge/api.py b/src/firefly_forge/api.py index 9168d73aa..56b636ef2 100644 --- a/src/firefly_forge/api.py +++ b/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,6 +811,7 @@ def coordinator( routing: TaskRouting, runtime_roles: RuntimeRoles, worker_evidence: WorkerEvidence, + stage_activity: EngineActivity, ) -> Engine: engine = Engine( store, @@ -804,6 +819,7 @@ def coordinator( 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, diff --git a/src/firefly_forge/engine.py b/src/firefly_forge/engine.py index 4354522a1..039251bc4 100644 --- a/src/firefly_forge/engine.py +++ b/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: diff --git a/src/firefly_forge/engine_activity.py b/src/firefly_forge/engine_activity.py new file mode 100644 index 000000000..18e0fc1bb --- /dev/null +++ b/src/firefly_forge/engine_activity.py @@ -0,0 +1,200 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Best-effort observations of locally owned Engine stages, separate from admission.""" + +from __future__ import annotations + +import asyncio +import logging +from contextvars import Context +from dataclasses import dataclass +from typing import Any +from uuid import UUID, uuid4 + +from pydantic import BaseModel, ConfigDict, Field, model_validator +from pyfly.container import service +from sqlalchemy import column, select, table + +from firefly_forge.engine_activity_persistence import SCOPE_FIELDS, ForgeEngineActivityRepository +from firefly_forge.persistence import ForgeJournalEntity, ForgePlanEntity, ForgeTaskEntity +from firefly_forge.store import Store, identifier + +log = logging.getLogger(__name__) + + +class EngineActivitySettings(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, allow_inf_nan=False) + heartbeat_seconds: float = Field(default=5, gt=0) + stale_seconds: float = Field(default=20, gt=0) + sample_timeout_seconds: float = Field(default=1, gt=0) + + @model_validator(mode="after") + def observation_window(self) -> EngineActivitySettings: + if self.stale_seconds <= self.heartbeat_seconds: + raise ValueError("Observation expiry must exceed the heartbeat interval") + return self + + +@dataclass(frozen=True) +class LocalStage: + row_id: UUID + run_id: str + key: str + task: asyncio.Task[Any] + ownership: object + + +@service +class EngineActivity: + def __init__( + self, + store: Store, + records: ForgeEngineActivityRepository, + settings: EngineActivitySettings | None = None, + ) -> None: + self.store, self.records = store, records + self.settings = settings or EngineActivitySettings() + self.epoch = uuid4().hex + self.scopes: dict[tuple[str, str], LocalStage] = {} + self.poller: asyncio.Task[None] | None = None + self.lock = asyncio.Lock() + + async def begin(self, run_id: str, task_id: str, key: str) -> None: + try: + task = asyncio.current_task() + if task is None: + return + async with asyncio.timeout(self.settings.sample_timeout_seconds), self.lock: + ownership = self.store.execution_ownership_token + await self.store.assert_ownership() + async with self.store._unit() as session: + run = await self.store.get_run(run_id) + actor = run.get("execution_actor") or {} + # Delegated SCM actors need their own current delegation projection. + if actor.get("authentication_method") not in {"local_bootstrap", "oidc_session", "oidc_bearer"}: + return + stage = await session.scalar( + select(ForgeTaskEntity).where( + ForgeTaskEntity.run_id == identifier(run_id), + ForgeTaskEntity.task_id == task_id, + ) + ) + if stage is None or stage.data.get("idempotency_key") != key: + return + intent = await session.scalar( + select(ForgeJournalEntity).where( + ForgeJournalEntity.run_id == identifier(run_id), + ForgeJournalEntity.task_id == task_id, + ForgeJournalEntity.idempotency_key == key, + ForgeJournalEntity.kind == "intent", + ) + ) + if intent is None: + return + plan = await session.scalar( + select(ForgePlanEntity) + .where( + ForgePlanEntity.run_id == identifier(run_id), + ) + .order_by(ForgePlanEntity.revision.desc()) + .limit(1) + ) + if plan is None: + return + backend = table("pg_stat_activity", column("pid"), column("backend_start"), schema="pg_catalog") + started = await session.scalar( + select(backend.c.backend_start).where( + backend.c.pid == self.store._ownership_pid, + ) + ) + if started is None or ownership is not self.store.execution_ownership_token: + return + row = await self.records.begin( + { + "run_id": identifier(run_id), + "actor_id": identifier(actor["user_id"]), + "project_id": identifier(run["project_id"]) if run.get("project_id") else None, + "plan_id": stage.plan_id, + "current_plan_id": plan.id, + "plan_digest": plan.digest, + "task_id": task_id, + "attempt": stage.attempt, + "invocation_key": key, + "intent_digest": intent.digest, + "owner_epoch": self.epoch, + "owner_pid": self.store._ownership_pid, + "owner_started_at": started, + "scope": { + "repository_id": identifier(run["repository_id"]).hex, + **{name: run.get(name) for name in SCOPE_FIELDS}, + }, + }, + self.settings.stale_seconds, + self.store.schema, + ) + if row is not None: + self.scopes[(run_id, key)] = LocalStage(row, run_id, key, task, ownership) + if row is not None and (self.poller is None or self.poller.done()): + self.poller = asyncio.create_task(self._poll(), context=Context()) + except Exception: + log.warning("Engine stage observation unavailable; recorded task state remains unchanged") + + async def end(self, run_id: str, key: str | None = None) -> None: + async with self.lock: + selected = [ + value for value in self.scopes.values() if value.run_id == run_id and (key is None or value.key == key) + ] + for value in selected: + self.scopes.pop((value.run_id, value.key), None) + await self._retire(selected) + + async def _retire(self, scopes: list[LocalStage]) -> None: + if not scopes: + return + try: + # End only our exact rows; SQL fence/CAS prevents retiring a replacement. + async with asyncio.timeout(self.settings.sample_timeout_seconds), self.store._unit(): + await self.records.end([value.row_id for value in scopes], self.epoch, self.store.schema) + except Exception: + log.warning("Engine stage observation could not end; its SQL-clock expiry remains in force") + + async def pulse(self) -> None: + async with self.lock: + ended = [value for value in self.scopes.values() if value.task.done() or value.task.cancelling()] + for value in ended: + self.scopes.pop((value.run_id, value.key), None) + await self._retire(ended) + if not self.scopes: + return + try: + async with asyncio.timeout(self.settings.sample_timeout_seconds): + ownership = self.store.execution_ownership_token + await self.store.assert_ownership() + selected = [value for value in self.scopes.values() if value.ownership is ownership] + async with self.store._unit(): + renewed = await self.records.renew( + [value.row_id for value in selected], + self.epoch, + self.settings.stale_seconds, + self.store.schema, + ) + kept = set(renewed) + self.scopes = {key: value for key, value in self.scopes.items() if value.row_id in kept} + except Exception: + # A failed sample cannot be used to restart or extend expired custody. + self.scopes.clear() + log.warning("Engine stage observation renewal unavailable; current activity becomes unknown") + + async def _poll(self) -> None: + while self.scopes: + await asyncio.sleep(self.settings.heartbeat_seconds) + await self.pulse() + + async def close(self) -> None: + if self.poller is not None: + self.poller.cancel() + await asyncio.gather(self.poller, return_exceptions=True) + self.poller = None + async with self.lock: + scopes = list(self.scopes.values()) + self.scopes.clear() + await self._retire(scopes) diff --git a/src/firefly_forge/engine_activity_persistence.py b/src/firefly_forge/engine_activity_persistence.py new file mode 100644 index 000000000..f8f51e971 --- /dev/null +++ b/src/firefly_forge/engine_activity_persistence.py @@ -0,0 +1,288 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Factory-local, expiring Engine-stage observations; never execution authority.""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from types import SimpleNamespace +from typing import Any +from typing import cast as typing_cast +from uuid import UUID, uuid4 + +from pyfly.container import repository +from pyfly.data.relational.sqlalchemy import BaseEntity, Repository +from sqlalchemy import ( + DateTime, + ForeignKey, + String, + UniqueConstraint, + and_, + cast, + column, + func, + literal, + or_, + select, + table, + update, +) +from sqlalchemy.dialects.postgresql import JSONB, insert +from sqlalchemy.orm import Mapped, mapped_column + +from firefly_forge.access_persistence import ForgeMembershipEntity +from firefly_forge.identity_persistence import ForgeIdentityUserEntity +from firefly_forge.persistence import ForgeJournalEntity, ForgePlanEntity, ForgeRunEntity, ForgeTaskEntity + +SCOPE_FIELDS = ( + "project_id", + "component_snapshot", + "execution_actor", + "sandbox_snapshot", + "policy_snapshot", + "role_snapshot", + "work_duration", + "config_revision", + "knowledge_context_digest", + "agent", + "assurance", + "commands", + "tool_policy_snapshot", + "mcp_snapshot", + "attempt", +) + + +class ForgeEngineActivityEntity(BaseEntity): + __tablename__ = "forge_engine_activity" + __table_args__ = (UniqueConstraint("run_id", "task_id"),) + run_id: Mapped[UUID] = mapped_column(ForeignKey("forge_runs.id"), index=True) + actor_id: Mapped[UUID] = mapped_column(ForeignKey("forge_identity_users.id"), index=True) + project_id: Mapped[UUID | None] = mapped_column(ForeignKey("forge_projects.id"), nullable=True, index=True) + plan_id: Mapped[UUID] = mapped_column(ForeignKey("forge_plans.id")) + current_plan_id: Mapped[UUID] = mapped_column(ForeignKey("forge_plans.id")) + plan_digest: Mapped[str] = mapped_column(String(64)) + task_id: Mapped[str] = mapped_column(String(100)) + attempt: Mapped[int] = mapped_column() + intent_digest: Mapped[str] = mapped_column(String(64)) + invocation_key: Mapped[str] = mapped_column(String(64)) + owner_epoch: Mapped[str] = mapped_column(String(32)) + owner_pid: Mapped[int] = mapped_column() + owner_started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + status: Mapped[str] = mapped_column(String(16)) + scope: Mapped[dict[str, Any]] = mapped_column(JSONB) + observed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + + +def scope_expression(run: Any = ForgeRunEntity) -> Any: + return func.jsonb_build_object( + "repository_id", + func.replace(cast(run.repository_id, String), "-", ""), + *[part for key in SCOPE_FIELDS for part in (key, run.data[key])], + ) + + +def factory_fence_predicate(schema: str, pid: Any, started: Any) -> Any: + """A session lock is necessary, but cannot by itself prove any run is active.""" + backend = table("pg_stat_activity", column("pid"), column("backend_start"), schema="pg_catalog") + locks = table( + "pg_locks", + *(column(key) for key in ("pid", "locktype", "granted", "objsubid", "classid", "objid")), + schema="pg_catalog", + ) + key = func.hashtextextended("forge:execution:" + schema, 0) + held = ( + select(literal(1)) + .select_from(locks) + .where( + locks.c.pid == pid, + locks.c.locktype == "advisory", + locks.c.granted.is_(True), + locks.c.objsubid == 1, + cast(locks.c.classid, String) == cast(key.op(">>")(32).op("&")(4294967295), String), + cast(locks.c.objid, String) == cast(key.op("&")(4294967295), String), + ) + .correlate_except(locks) + .exists() + ) + return ( + select(literal(1)) + .select_from(backend) + .where( + backend.c.pid == pid, + backend.c.backend_start == started, + held, + ) + .correlate_except(backend) + .exists() + ) + + +def actor_predicate(run: Any = ForgeRunEntity, activity: Any = ForgeEngineActivityEntity) -> Any: + """Only the explicitly supported canonical local/OIDC actor paths are observed.""" + user, membership = ForgeIdentityUserEntity, ForgeMembershipEntity + actor = run.data["execution_actor"] + method = actor["authentication_method"].as_string() + admin = and_( + actor["global_admin"].as_boolean().is_(True), + actor["api_audience"].as_string() == user.api_roles_audience, + user.api_roles_issued_at.is_not(None), + user.api_roles_expires_at > func.clock_timestamp(), + user.api_roles.contains(["forge-admin"]), + ) + member = ( + select(membership.id) + .where( + membership.project_id == activity.project_id, + membership.user_id == user.id, + membership.active.is_(True), + membership.role.in_(("owner", "maintainer", "developer")), + ) + .correlate_except(membership) + .exists() + ) + isolated = and_( + run.data["agent"].as_string().not_in(("codex", "claude")), + func.coalesce(run.data["assurance"]["review_agent"].as_string(), "").not_in(("codex", "claude")), + func.coalesce(run.data["sandbox_snapshot"]["mode"].as_string(), "") != "host", + ) + return ( + select(user.id) + .where( + func.replace(cast(user.id, String), "-", "") == actor["user_id"].as_string(), + user.id == activity.actor_id, + user.enabled.is_(True), + or_( + method == "local_bootstrap", + and_(method.in_(("oidc_session", "oidc_bearer")), or_(admin, and_(member, isolated))), + ), + ) + .correlate_except(user) + .exists() + ) + + +def current_activity_predicate( + schema: str, + run: Any = ForgeRunEntity, + task: Any = ForgeTaskEntity, + activity: Any = ForgeEngineActivityEntity, +) -> Any: + latest = ( + select(func.max(ForgePlanEntity.revision)) + .where(ForgePlanEntity.run_id == run.id) + .correlate_except(ForgePlanEntity) + .scalar_subquery() + ) + plan = ( + select(ForgePlanEntity.id) + .where( + ForgePlanEntity.run_id == run.id, + ForgePlanEntity.id == activity.current_plan_id, + ForgePlanEntity.digest == activity.plan_digest, + ForgePlanEntity.revision == latest, + ) + .correlate_except(ForgePlanEntity) + .exists() + ) + return and_( + activity.run_id == run.id, + task.run_id == run.id, + activity.task_id == task.task_id, + activity.plan_id == task.plan_id, + activity.attempt == task.attempt, + activity.invocation_key == task.data["idempotency_key"].as_string(), + activity.status == "active", + activity.expires_at > func.clock_timestamp(), + task.status == "running", + run.data["active_task"].as_string() == task.task_id, + run.status.in_(("preparing", "developing", "validating", "running")), + func.coalesce(run.data["cancel_request"]["status"].as_string(), "") != "requested", + activity.scope == scope_expression(run), + func.coalesce(func.replace(cast(activity.project_id, String), "-", ""), "") + == func.coalesce(run.data["project_id"].as_string(), ""), + plan, + actor_predicate(run, activity), + select(ForgeJournalEntity.id) + .where( + ForgeJournalEntity.run_id == run.id, + ForgeJournalEntity.task_id == task.task_id, + ForgeJournalEntity.idempotency_key == activity.invocation_key, + ForgeJournalEntity.kind == "intent", + ForgeJournalEntity.digest == activity.intent_digest, + ) + .correlate_except(ForgeJournalEntity) + .exists(), + factory_fence_predicate(schema, activity.owner_pid, activity.owner_started_at), + ) + + +@repository +class ForgeEngineActivityRepository(Repository[ForgeEngineActivityEntity, UUID]): + async def begin(self, values: dict[str, Any], seconds: float, schema: str) -> UUID | None: + current = ForgeEngineActivityEntity + stamp = func.clock_timestamp() + statement = insert(current).from_select( + ["id", *values, "status", "observed_at", "expires_at", "created_at", "updated_at"], + select( + literal(uuid4()), + *[literal(value, type_=current.__table__.c[key].type) for key, value in values.items()], + literal("active"), + stamp, + stamp + timedelta(seconds=seconds), + stamp, + stamp, + ) + .select_from(ForgeRunEntity) + .join(ForgeTaskEntity, ForgeTaskEntity.run_id == ForgeRunEntity.id) + .where( + ForgeRunEntity.id == values["run_id"], + current_activity_predicate( + schema, + activity=SimpleNamespace( + **{key: literal(value, type_=current.__table__.c[key].type) for key, value in values.items()}, + status=literal("active"), + expires_at=stamp + timedelta(seconds=seconds), + ), + ), + ), + ) + returning = statement.on_conflict_do_update( + index_elements=[current.run_id, current.task_id], + set_={key: statement.excluded[key] for key in (*values, "id", "status", "observed_at", "expires_at")}, + where=or_(current.owner_epoch != values["owner_epoch"], current.invocation_key != values["invocation_key"]), + ).returning(current.id) + return typing_cast(UUID | None, await self._require_session().scalar(returning)) + + async def renew(self, identifiers: list[UUID], epoch: str, seconds: float, schema: str) -> list[UUID]: + row = ForgeEngineActivityEntity + stamp = func.clock_timestamp() + return list( + await self._require_session().scalars( + update(row) + .where( + row.id.in_(identifiers), + row.owner_epoch == epoch, + select(ForgeRunEntity.id) + .join(ForgeTaskEntity, ForgeTaskEntity.run_id == ForgeRunEntity.id) + .where(current_activity_predicate(schema)) + .correlate(row) + .exists(), + ) + .values(observed_at=stamp, expires_at=stamp + timedelta(seconds=seconds)) + .returning(row.id) + ) + ) + + async def end(self, identifiers: list[UUID], epoch: str, schema: str) -> None: + row = ForgeEngineActivityEntity + await self._require_session().execute( + update(row) + .where( + row.id.in_(identifiers), + row.owner_epoch == epoch, + row.status == "active", + factory_fence_predicate(schema, row.owner_pid, row.owner_started_at), + ) + .values(status="ended", expires_at=func.clock_timestamp()) + ) diff --git a/src/firefly_forge/factory_dashboard.py b/src/firefly_forge/factory_dashboard.py index 878f765e2..8616fe638 100644 --- a/src/firefly_forge/factory_dashboard.py +++ b/src/firefly_forge/factory_dashboard.py @@ -6,6 +6,7 @@ import re from collections.abc import AsyncIterator from contextlib import asynccontextmanager +from datetime import UTC, datetime from typing import Any from uuid import UUID @@ -160,6 +161,7 @@ async def observe( after=int(position) if position else None, limit=page.limit, bucket=page.bucket, + schema=self.store.schema, allowed_projects=allowed_projects, ) if scope.kind == "run" and not sum(raw["counts"].values()): @@ -208,10 +210,21 @@ async def observe( } unmapped = 0 for record in raw["stages"]: + state, deadline = record["state"], record.get("observed_until") + if state == "active" and not ( + isinstance(deadline, datetime) and deadline.tzinfo is not None and deadline > observed_at + ): + state = "unknown" if record["stage"] in stages: - stages[record["stage"]][record["state"]] += record["count"] - stages[record["stage"]]["recorded_running"] += record.get("recorded_running", 0) - elif record["state"] == "active": + stage_record = stages[record["stage"]] + stage_record[state] += record["count"] + stage_record["recorded_running"] += record.get("recorded_running", 0) + if state == "active": + expiry = deadline.astimezone(UTC).isoformat() + stage_record["observed_until"] = ( + min(stage_record["observed_until"], expiry) if stage_record["observed_until"] else expiry + ) + elif state == "active": unmapped += record["count"] rows = raw["rows"] return { diff --git a/src/firefly_forge/factory_dashboard_persistence.py b/src/firefly_forge/factory_dashboard_persistence.py index 0fdb54ac5..3d528ab19 100644 --- a/src/firefly_forge/factory_dashboard_persistence.py +++ b/src/firefly_forge/factory_dashboard_persistence.py @@ -10,8 +10,10 @@ from pyfly.data.relational.sqlalchemy import Repository from sqlalchemy import BigInteger, ColumnElement, Numeric, Select, String, and_, case, cast, func, literal, select from sqlalchemy.dialects.postgresql import BIT, JSONB +from sqlalchemy.orm import aliased from firefly_forge.access_persistence import ForgeMembershipEntity +from firefly_forge.engine_activity_persistence import ForgeEngineActivityEntity, current_activity_predicate from firefly_forge.factory_dashboard_models import RUN_BUCKETS, DashboardScope from firefly_forge.identity import Principal from firefly_forge.identity_persistence import ForgeIdentitySessionEntity, ForgeIdentityUserEntity @@ -257,6 +259,7 @@ async def snapshot( after: int | None, limit: int, bucket: str | None, + schema: str, allowed_projects: tuple[str, ...] | None = None, ) -> dict[str, Any]: session = self._require_session() @@ -284,9 +287,11 @@ async def snapshot( ).subquery() tasks = ForgeTaskEntity stage = case((tasks.task_id.like("publish%"), literal("publish")), else_=tasks.task_id) - # The controller ledger lives in its separate private schema. A running task - # is retained progress, not a per-run heartbeat; never infer current motion. + activity = ForgeEngineActivityEntity + observed_run = aliased(ForgeRunEntity, scoped) + observed = current_activity_predicate(schema, run=observed_run, task=tasks, activity=activity) task_state = case( + (observed, literal("active")), (tasks.status.in_(("done", "succeeded")), literal("done")), (tasks.status.in_(("queued", "pending", "waiting")), literal("waiting")), (tasks.status == "skipped", literal("skipped")), @@ -300,9 +305,11 @@ async def snapshot( task_state.label("state"), func.count().label("count"), func.count().filter(tasks.status == "running").label("recorded_running"), + func.min(case((observed, activity.expires_at), else_=None)).label("observed_until"), ) .select_from(tasks) .join(scoped, scoped.c.id == tasks.run_id) + .outerjoin(activity, and_(activity.run_id == tasks.run_id, activity.task_id == tasks.task_id)) .join(latest_plans, and_(latest_plans.c.run_id == tasks.run_id, latest_plans.c.position == 1)) .group_by(stage, task_state) ) diff --git a/src/firefly_forge/migrations.py b/src/firefly_forge/migrations.py index a7807408d..333402d18 100644 --- a/src/firefly_forge/migrations.py +++ b/src/firefly_forge/migrations.py @@ -35,6 +35,7 @@ OAuthGrantEntity, OAuthRegistrationEntity, ) +from firefly_forge.engine_activity_persistence import ForgeEngineActivityEntity from firefly_forge.execution_request_persistence import ForgeExecutionRequestEntity, ForgeExecutionRequestEventEntity from firefly_forge.framework_knowledge_persistence import FrameworkKnowledgeRecord from firefly_forge.identity_persistence import ( @@ -86,7 +87,7 @@ ForgeIdentityAdminOperationEntity, ) -SCHEMA_REVISION = 32 +SCHEMA_REVISION = 33 async def migrate(session: AsyncSession, schema: str) -> None: @@ -122,6 +123,7 @@ async def migrate(session: AsyncSession, schema: str) -> None: ForgeDeploymentEntity, ForgePlanEntity, ForgeTaskEntity, + ForgeEngineActivityEntity, ForgeJournalEntity, ForgeObservationEntity, ForgeIncidentEntity, diff --git a/src/firefly_forge/source_budget_allowances.py b/src/firefly_forge/source_budget_allowances.py index d1f32d5ef..000414e06 100644 --- a/src/firefly_forge/source_budget_allowances.py +++ b/src/firefly_forge/source_budget_allowances.py @@ -399,6 +399,8 @@ def admission_contract_digest() -> str: "identity_persistence.py", "access_persistence.py", "product_provenance.py", + "engine_activity.py", + "engine_activity_persistence.py", ) } ) diff --git a/tests/fixtures/engine_activity/schema32_migrations.py b/tests/fixtures/engine_activity/schema32_migrations.py new file mode 100644 index 000000000..a7807408d --- /dev/null +++ b/tests/fixtures/engine_activity/schema32_migrations.py @@ -0,0 +1,463 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Versioned, additive startup migration for Forge's own PostgreSQL tables.""" + +from __future__ import annotations + +import os + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from firefly_forge.access_persistence import ForgeAccessEventEntity, ForgeMembershipEntity +from firefly_forge.activity_persistence import ForgeEventContentEntity +from firefly_forge.analytics_persistence import ForgeAnalyticsBindingEntity, ForgeAnalyticsConfigurationReceiptEntity +from firefly_forge.application_persistence import ForgeIncidentEntity, ForgeObservationEntity +from firefly_forge.assistant_attachment_persistence import ForgeAssistantAttachmentEntity +from firefly_forge.assistant_persistence import ( + ForgeAssistantEventEntity, + ForgeAssistantSessionEntity, + ForgeAssistantTurnEntity, +) +from firefly_forge.campaign_delivery_plan_persistence import ForgeCampaignDeliveryPlanEntity +from firefly_forge.campaign_report_history_persistence import ForgeCampaignReportArchiveEntity +from firefly_forge.collaboration_persistence import ( + ForgeChannelInboxEntity, + ForgeCollaborationSettingsEntity, + ForgeProductDefinitionEntity, + ForgeWebhookInboxEntity, +) +from firefly_forge.connection_oauth_persistence import ( + OAuthAttemptEntity, + OAuthConnectionEntity, + OAuthCredentialEvidenceEntity, + OAuthEffectEntity, + OAuthEventEntity, + OAuthGrantEntity, + OAuthRegistrationEntity, +) +from firefly_forge.execution_request_persistence import ForgeExecutionRequestEntity, ForgeExecutionRequestEventEntity +from firefly_forge.framework_knowledge_persistence import FrameworkKnowledgeRecord +from firefly_forge.identity_persistence import ( + ForgeIdentityAttemptEntity, + ForgeIdentitySessionEntity, + ForgeIdentityUserEntity, +) +from firefly_forge.knowledge_persistence import ForgeKnowledgeUploadEntity +from firefly_forge.learning_observation_attempt_persistence import ( + ForgeLearningObservationAttemptEntity, + observation_attempt_guards, +) +from firefly_forge.learning_persistence import ( + ForgeLearningEventEntity, + ForgeLearningProjectionEntity, + ForgeLearningRecordEntity, +) +from firefly_forge.oauth_provider_policy_persistence import ( + OAuthProviderPolicyEntity, + OAuthProviderPolicyOperationEntity, + OAuthProviderPolicyRevisionEntity, +) +from firefly_forge.persistence import ( + ForgeDeploymentEntity, + ForgeEventEntity, + ForgeJournalEntity, + ForgePlanEntity, + ForgeRepositoryEntity, + ForgeRunEntity, + ForgeTaskEntity, +) +from firefly_forge.personal_token_persistence import ForgePersonalTokenEntity, ForgePersonalTokenEventEntity +from firefly_forge.project_campaign_persistence import ( + ForgeCampaignChildEntity, + ForgeCampaignEntity, + ForgeCampaignEventEntity, +) +from firefly_forge.project_persistence import ForgeAnalysisEntity, ForgeProjectEntity, ForgeProjectRevisionEntity +from firefly_forge.run_creation_persistence import ( + ForgeRunCreationCancellationEntity, + ForgeRunCreationRequestEntity, + ForgeRunCreationReservationEntity, +) +from firefly_forge.runtime_role_persistence import ForgeRuntimeTopologyEntity +from firefly_forge.sandbox_persistence import ForgeSandboxEntity +from firefly_forge.tool_persistence import ForgeToolInvocationEntity +from firefly_forge.user_administration_persistence import ( + ForgeIdentityAdminEventEntity, + ForgeIdentityAdminOperationEntity, +) + +SCHEMA_REVISION = 32 + + +async def migrate(session: AsyncSession, schema: str) -> None: + await session.execute( + text("SELECT pg_advisory_xact_lock(hashtextextended(:key, 0))"), {"key": "forge:migration:" + schema} + ) + await session.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{schema}"')) + await session.execute(text(f'SET LOCAL search_path TO "{schema}"')) + await session.execute( + text( + "CREATE TABLE IF NOT EXISTS forge_schema_migrations " + "(revision INTEGER PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())" + ) + ) + version = await session.scalar(text("SELECT max(revision) FROM forge_schema_migrations")) + if version is not None and version > SCHEMA_REVISION: + raise RuntimeError("The PostgreSQL schema requires a newer Forge release") + if version is None or version < SCHEMA_REVISION: + connection = await session.connection() + tables = [ + ForgeRunEntity.metadata.tables[entity.__tablename__] + for entity in ( + ForgeRuntimeTopologyEntity, + ForgeExecutionRequestEntity, + ForgeExecutionRequestEventEntity, + ForgeRunCreationRequestEntity, + ForgeRunCreationReservationEntity, + ForgeRunCreationCancellationEntity, + ForgeRepositoryEntity, + ForgeRunEntity, + ForgeEventEntity, + ForgeEventContentEntity, + ForgeDeploymentEntity, + ForgePlanEntity, + ForgeTaskEntity, + ForgeJournalEntity, + ForgeObservationEntity, + ForgeIncidentEntity, + ForgeCollaborationSettingsEntity, + ForgeWebhookInboxEntity, + ForgeProductDefinitionEntity, + ForgeChannelInboxEntity, + ForgeKnowledgeUploadEntity, + FrameworkKnowledgeRecord, + ForgeProjectEntity, + ForgeAnalyticsBindingEntity, + ForgeAnalyticsConfigurationReceiptEntity, + ForgeProjectRevisionEntity, + ForgeAnalysisEntity, + ForgeAssistantSessionEntity, + ForgeAssistantTurnEntity, + ForgeAssistantEventEntity, + ForgeAssistantAttachmentEntity, + ForgeSandboxEntity, + ForgeCampaignEntity, + ForgeCampaignChildEntity, + ForgeCampaignEventEntity, + ForgeCampaignReportArchiveEntity, + ForgeCampaignDeliveryPlanEntity, + ForgeIdentityUserEntity, + ForgeIdentitySessionEntity, + ForgeIdentityAttemptEntity, + ForgePersonalTokenEntity, + ForgePersonalTokenEventEntity, + ForgeMembershipEntity, + ForgeAccessEventEntity, + ForgeIdentityAdminOperationEntity, + ForgeIdentityAdminEventEntity, + ForgeToolInvocationEntity, + ForgeLearningRecordEntity, + ForgeLearningEventEntity, + ForgeLearningProjectionEntity, + ForgeLearningObservationAttemptEntity, + OAuthConnectionEntity, + OAuthRegistrationEntity, + OAuthAttemptEntity, + OAuthGrantEntity, + OAuthEffectEntity, + OAuthEventEntity, + OAuthCredentialEvidenceEntity, + OAuthProviderPolicyEntity, + OAuthProviderPolicyRevisionEntity, + OAuthProviderPolicyOperationEntity, + ) + ] + await connection.run_sync(lambda sync: ForgeRunEntity.metadata.create_all(sync, tables=tables)) + await session.execute( + text("ALTER TABLE forge_identity_users ADD COLUMN IF NOT EXISTS contact_email VARCHAR(320)") + ) + await session.execute( + text( + "ALTER TABLE forge_identity_users " + "ADD COLUMN IF NOT EXISTS api_roles JSONB NOT NULL DEFAULT '[]'::jsonb, " + "ADD COLUMN IF NOT EXISTS api_roles_audience VARCHAR(2048), " + "ADD COLUMN IF NOT EXISTS api_roles_issued_at DOUBLE PRECISION, " + "ADD COLUMN IF NOT EXISTS api_roles_expires_at TIMESTAMPTZ" + ) + ) + await session.execute( + text( + "ALTER TABLE forge_assistant_turns " + "ADD COLUMN IF NOT EXISTS owner_id VARCHAR(32), " + "ADD COLUMN IF NOT EXISTS claim_epoch BIGINT NOT NULL DEFAULT 0, " + "ADD COLUMN IF NOT EXISTS lease_expires_at TIMESTAMPTZ, " + "ADD COLUMN IF NOT EXISTS last_heartbeat_at TIMESTAMPTZ" + ) + ) + issuer = os.environ.get("FORGE_OIDC_ISSUER") + if issuer: + await session.execute( + text( + "UPDATE forge_identity_users AS u SET " + "contact_email=p.data #>> '{request,email}', " + "display_name=COALESCE(NULLIF(u.display_name,''), " + "NULLIF(left(trim(concat_ws(' ',p.data #>> '{request,first_name}', " + "p.data #>> '{request,last_name}')),200),''), " + "left(p.data #>> '{request,username}',200)) " + "FROM forge_identity_admin_operations AS p " + "WHERE u.issuer=:issuer AND u.contact_email IS NULL " + "AND p.kind='provision' AND p.status='completed' " + "AND p.data->>'user_id'=replace(u.id::text,'-','') " + "AND p.data->>'subject'=u.subject " + "AND length(p.data #>> '{request,email}') BETWEEN 3 AND 320 " + "AND p.data #>> '{request,email}' ~ '^[^@[:space:]]+@[^@[:space:]]+[.][^@[:space:]]+$' " + "AND EXISTS (SELECT 1 FROM forge_identity_admin_events AS e " + "WHERE e.operation_id=p.id AND e.data->>'kind'='remote_receipt' " + "AND e.data #>> '{payload,status}'='completed' " + "AND e.data #>> '{payload,user_id}'=replace(u.id::text,'-','') " + "AND e.data #>> '{payload,subject}'=u.subject)" + ), + {"issuer": issuer}, + ) + await session.execute( + text( + "CREATE OR REPLACE FUNCTION forge_refuse_immutable_change() " + "RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN " + "RAISE EXCEPTION 'Forge execution history is immutable'; END; $$" + ) + ) + for table in ( + "forge_runtime_topology", + "forge_execution_request_events", + "forge_run_creation_requests", + "forge_run_creation_reservations", + "forge_run_creation_cancellations", + "forge_plans", + "forge_journal", + "forge_observations", + "forge_product_definitions", + "forge_project_revisions", + "forge_analytics_configuration_receipts", + "forge_assistant_events", + "forge_assistant_attachments", + "forge_event_content", + "forge_campaign_children", + "forge_campaign_events", + "forge_campaign_report_archives", + "forge_campaign_delivery_plan_revisions", + "forge_access_events", + "forge_identity_admin_events", + "forge_learning_events", + ): + await session.execute( + text( + f"CREATE OR REPLACE TRIGGER {table}_immutable " + f"BEFORE UPDATE OR DELETE ON {table} FOR EACH ROW " + "EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) + await session.execute( + text(""" + CREATE OR REPLACE FUNCTION forge_guard_execution_request() RETURNS trigger AS $$ + BEGIN + IF OLD.run_id IS DISTINCT FROM NEW.run_id OR OLD.kind IS DISTINCT FROM NEW.kind + OR OLD.request_key IS DISTINCT FROM NEW.request_key + OR OLD.data->'request' IS DISTINCT FROM NEW.data->'request' + OR OLD.data->'request_digest' IS DISTINCT FROM NEW.data->'request_digest' THEN + RAISE EXCEPTION 'Execution request admission is immutable'; + END IF; + RETURN NEW; + END; $$ LANGUAGE plpgsql + """) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_execution_request_admission " + "BEFORE UPDATE ON forge_execution_requests FOR EACH ROW " + "EXECUTE FUNCTION forge_guard_execution_request()" + ) + ) + await session.execute( + text("INSERT INTO forge_schema_migrations(revision) VALUES (:revision)"), {"revision": SCHEMA_REVISION} + ) + await session.execute( + text(""" + CREATE OR REPLACE FUNCTION forge_guard_tool_invocation() RETURNS trigger AS $$ + BEGIN + IF OLD.run_id IS DISTINCT FROM NEW.run_id OR OLD.task_id IS DISTINCT FROM NEW.task_id + OR OLD.tool IS DISTINCT FROM NEW.tool OR OLD.correlation_id IS DISTINCT FROM NEW.correlation_id + OR OLD.data->'admission' IS DISTINCT FROM NEW.data->'admission' + OR OLD.data->'request_digest' IS DISTINCT FROM NEW.data->'request_digest' + OR (OLD.status IN ('succeeded','rejected','abandoned') + AND (OLD.status IS DISTINCT FROM NEW.status OR OLD.data IS DISTINCT FROM NEW.data)) THEN + RAISE EXCEPTION 'Tool invocation admission and completed outcome are immutable'; + END IF; + RETURN NEW; + END; $$ LANGUAGE plpgsql + """) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_tool_invocation_admission " + "BEFORE UPDATE ON forge_tool_invocations FOR EACH ROW EXECUTE FUNCTION forge_guard_tool_invocation()" + ) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_tool_invocation_delete " + "BEFORE DELETE ON forge_tool_invocations FOR EACH ROW EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) + + await session.execute( + text(""" + CREATE OR REPLACE FUNCTION forge_guard_learning_record() RETURNS trigger AS $$ + BEGIN + IF OLD.run_id IS DISTINCT FROM NEW.run_id OR OLD.repository_id IS DISTINCT FROM NEW.repository_id + OR OLD.correlation_id IS DISTINCT FROM NEW.correlation_id OR OLD.kind IS DISTINCT FROM NEW.kind + OR OLD.data IS DISTINCT FROM NEW.data THEN + RAISE EXCEPTION 'Learning origin and evidence are immutable'; + END IF; + IF OLD.state IS DISTINCT FROM NEW.state AND NOT ( + OLD.state IN ('candidate','verified') AND NEW.state IN ('invalidated','superseded') + ) THEN RAISE EXCEPTION 'Learning state transition requires native evidence'; END IF; + RETURN NEW; + END; $$ LANGUAGE plpgsql + """) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_learning_admission BEFORE UPDATE ON forge_learning_records " + "FOR EACH ROW EXECUTE FUNCTION forge_guard_learning_record()" + ) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_learning_delete BEFORE DELETE ON forge_learning_records " + "FOR EACH ROW EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) + + await session.execute( + text(""" + CREATE OR REPLACE FUNCTION forge_guard_learning_projection() RETURNS trigger AS $$ + BEGIN + IF OLD.record_id IS DISTINCT FROM NEW.record_id + OR OLD.payload_digest IS DISTINCT FROM NEW.payload_digest + OR OLD.data->'payload' IS DISTINCT FROM NEW.data->'payload' + OR (OLD.data ? 'target' AND (OLD.data->'target' IS DISTINCT FROM NEW.data->'target' + OR OLD.data->'workspace_id' IS DISTINCT FROM NEW.data->'workspace_id' + OR OLD.data->'content_sha256' IS DISTINCT FROM NEW.data->'content_sha256' + OR OLD.data->'request_id' IS DISTINCT FROM NEW.data->'request_id')) + OR (OLD.state IN ('verified','abandoned') AND + (OLD.state IS DISTINCT FROM NEW.state OR OLD.data IS DISTINCT FROM NEW.data)) THEN + RAISE EXCEPTION 'Learning projection admission and final outcome are immutable'; + END IF; + IF OLD.state IS DISTINCT FROM NEW.state AND NOT ( + (OLD.state='pending' AND NEW.state='executing') OR + (OLD.state='executing' AND NEW.state IN ('source_admitted','unknown','verified')) OR + (OLD.state IN ('source_admitted','unknown') AND NEW.state IN ('verified','abandoned')) + ) THEN RAISE EXCEPTION 'Learning projection transition cannot replay a submission'; END IF; + RETURN NEW; + END; $$ LANGUAGE plpgsql + """) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_learning_projection_admission " + "BEFORE UPDATE ON forge_learning_projections FOR EACH ROW " + "EXECUTE FUNCTION forge_guard_learning_projection()" + ) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_learning_projection_delete " + "BEFORE DELETE ON forge_learning_projections FOR EACH ROW " + "EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) + + await _oauth_history_guards(session) + await _oauth_provider_policy_guards(session) + await observation_attempt_guards(session) + + +async def _oauth_history_guards(session: AsyncSession) -> None: + for table in ("forge_oauth_registrations", "forge_oauth_events", "forge_oauth_credential_evidence"): + await session.execute( + text( + f"CREATE OR REPLACE TRIGGER {table}_immutable " + f"BEFORE UPDATE OR DELETE ON {table} FOR EACH ROW " + "EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) + await session.execute( + text(""" + CREATE OR REPLACE FUNCTION forge_guard_oauth_effect() RETURNS trigger AS $$ + BEGIN + IF OLD.id IS DISTINCT FROM NEW.id OR OLD.connection_id IS DISTINCT FROM NEW.connection_id + OR OLD.request_id IS DISTINCT FROM NEW.request_id OR OLD.kind IS DISTINCT FROM NEW.kind + OR OLD.created_at IS DISTINCT FROM NEW.created_at + OR OLD.lease_expires_at IS DISTINCT FROM NEW.lease_expires_at + OR (OLD.data - ARRAY['result','observed_at','disposition']) IS DISTINCT FROM + (NEW.data - ARRAY['result','observed_at','disposition']) + OR (OLD.state IN ('completed','rejected','unknown') AND + (OLD.state IS DISTINCT FROM NEW.state OR OLD.data IS DISTINCT FROM NEW.data)) THEN + RAISE EXCEPTION 'OAuth effect admission and closed outcome are immutable'; + END IF; + IF OLD.state IS DISTINCT FROM NEW.state AND NOT ( + OLD.state='executing' AND NEW.state IN ('completed','rejected','unknown') + ) THEN RAISE EXCEPTION 'OAuth effect transition cannot replay a request'; END IF; + RETURN NEW; + END; $$ LANGUAGE plpgsql + """) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_oauth_effect_admission " + "BEFORE UPDATE ON forge_oauth_effects FOR EACH ROW EXECUTE FUNCTION forge_guard_oauth_effect()" + ) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_oauth_effect_delete " + "BEFORE DELETE ON forge_oauth_effects FOR EACH ROW EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) + + +async def _oauth_provider_policy_guards(session: AsyncSession) -> None: + for table in ("forge_oauth_provider_policy_revisions", "forge_oauth_provider_policy_operations"): + await session.execute( + text( + f"CREATE OR REPLACE TRIGGER {table}_immutable " + f"BEFORE UPDATE OR DELETE ON {table} FOR EACH ROW " + "EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) + await session.execute( + text(""" + CREATE OR REPLACE FUNCTION forge_guard_oauth_provider_policy() RETURNS trigger AS $$ + BEGIN + IF OLD.id IS DISTINCT FROM NEW.id OR OLD.name IS DISTINCT FROM NEW.name + OR NEW.revision <> OLD.revision + 1 THEN + RAISE EXCEPTION 'Provider policy identity is immutable and revisions must advance'; + END IF; + RETURN NEW; + END; $$ LANGUAGE plpgsql + """) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_oauth_provider_policy_revision " + "BEFORE UPDATE ON forge_oauth_provider_policies FOR EACH ROW " + "EXECUTE FUNCTION forge_guard_oauth_provider_policy()" + ) + ) + await session.execute( + text( + "CREATE OR REPLACE TRIGGER forge_oauth_provider_policy_delete " + "BEFORE DELETE ON forge_oauth_provider_policies FOR EACH ROW " + "EXECUTE FUNCTION forge_refuse_immutable_change()" + ) + ) diff --git a/tests/test_activity_archive_native.py b/tests/test_activity_archive_native.py index 308cca3ac..8a5a992ff 100644 --- a/tests/test_activity_archive_native.py +++ b/tests/test_activity_archive_native.py @@ -124,7 +124,7 @@ async def test_native_populated_schema29_upgrade_is_additive_and_does_not_backfi assert spec and spec.loader legacy = importlib.util.module_from_spec(spec) spec.loader.exec_module(legacy) - assert legacy.SCHEMA_REVISION == 29 and SCHEMA_REVISION == 32 + assert legacy.SCHEMA_REVISION == 29 monkeypatch.setattr(module, "migrate", legacy.migrate) store = await store_factory(tmp_path / "schema29/factory.db") from firefly_forge.models import RepositoryInput, RunInput diff --git a/tests/test_analytics_upgrade_native.py b/tests/test_analytics_upgrade_native.py index bdee25d4c..a483fdca7 100644 --- a/tests/test_analytics_upgrade_native.py +++ b/tests/test_analytics_upgrade_native.py @@ -53,7 +53,7 @@ async def test_populated_actual_schema28_to29_preserves_authority_and_receipts_t import firefly_forge.store as store_module legacy = previous_module("schema28_migrations", "05afcf7c9d6f792206fccd8505e1d24f4d828b9f0d171062ac2b6bb98274b021") - assert legacy.SCHEMA_REVISION == 28 and SCHEMA_REVISION == 32 + assert legacy.SCHEMA_REVISION == 28 monkeypatch.setattr(store_module, "migrate", legacy.migrate) store = await store_factory(tmp_path / "schema28" / "factory.db") async with store._unit() as session: diff --git a/tests/test_campaign_delivery_plans.py b/tests/test_campaign_delivery_plans.py index c9b2bd9b7..b3a77bdbe 100644 --- a/tests/test_campaign_delivery_plans.py +++ b/tests/test_campaign_delivery_plans.py @@ -303,7 +303,6 @@ async def test_native_revision_quota_and_schema25_upgrade_preserve_archives(stor c = await prepared(store, tmp_path) try: - assert SCHEMA_REVISION == 32 async with store._unit() as session: assert await session.scalar(text("SELECT count(*) FROM forge_event_content")) == 0 await session.execute(text("DROP TABLE forge_campaign_delivery_plan_revisions")) diff --git a/tests/test_campaign_report_history.py b/tests/test_campaign_report_history.py index f8e4a56fc..6a890d91c 100644 --- a/tests/test_campaign_report_history.py +++ b/tests/test_campaign_report_history.py @@ -398,7 +398,6 @@ async def test_upgrade_from_revision24_adds_immutable_archive_without_changing_c project, manager, journal, campaign, token = await seeded(store, tmp_path) try: - assert SCHEMA_REVISION == 32 before = await manager.campaign(project["id"], campaign["id"]) async with store._unit() as session: assert await session.scalar(text("SELECT count(*) FROM forge_event_content")) == 0 diff --git a/tests/test_checkpoint_mode_adoption_host.py b/tests/test_checkpoint_mode_adoption_host.py index e8cac5224..51256431d 100644 --- a/tests/test_checkpoint_mode_adoption_host.py +++ b/tests/test_checkpoint_mode_adoption_host.py @@ -183,6 +183,7 @@ async def test_real_begin_task_claims_old_intent_once_without_new_attempt_or_fak await c.service.request(c.run_id, uuid4().hex, request(candidate)) prepared = await prepare_current_mode(c.journal, c.run_id) engine = object.__new__(Engine) + engine.stage_activity = None engine.store = c.store engine.journal = c.journal engine.assert_ownership = c.store.assert_ownership @@ -274,6 +275,7 @@ def guarded(*args, __original=original, **kwargs): await c.service.request(c.run_id, uuid4().hex, request(candidate)) prepared = await prepare_current_mode(c.journal, c.run_id) engine = object.__new__(Engine) + engine.stage_activity = None engine.store = c.store engine.journal = c.journal engine.assert_ownership = c.store.assert_ownership diff --git a/tests/test_engine_activity_host.py b/tests/test_engine_activity_host.py new file mode 100644 index 000000000..830b1c65e --- /dev/null +++ b/tests/test_engine_activity_host.py @@ -0,0 +1,419 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Stage observations follow actual Engine execution, never retained running status.""" + +import asyncio +from contextlib import asynccontextmanager +from contextvars import ContextVar +from datetime import UTC, datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from pydantic import ValidationError +from sqlalchemy.dialects import postgresql + +from firefly_forge.engine import RETAINED_EXECUTION_FENCES, Engine +from firefly_forge.engine_activity import EngineActivity, EngineActivitySettings +from firefly_forge.engine_activity_persistence import ForgeEngineActivityRepository +from firefly_forge.project_access import ProjectAccess + + +@pytest.mark.asyncio +async def test_engine_begins_only_a_new_native_stage_and_ends_after_receipt(monkeypatch, tmp_path): + engine = Engine.__new__(Engine) + engine.assert_ownership = AsyncMock() + engine.store = SimpleNamespace(get_run=AsyncMock(return_value={})) + engine.journal = SimpleNamespace( + ensure_plan=AsyncMock(return_value={"tasks": [{"id": "implement", "attempt": 1, "status": "queued"}]}), + intent=AsyncMock(return_value={"disposition": "execute", "idempotency_key": "a" * 32}), + receipt=AsyncMock(), + ) + engine.stage_activity = SimpleNamespace(begin=AsyncMock(), end=AsyncMock()) + monkeypatch.setattr(ProjectAccess, "authorize_execution", AsyncMock()) + await engine._begin_task("b" * 32, "implement", tmp_path, {}) + engine.stage_activity.begin.assert_awaited_once_with("b" * 32, "implement", "a" * 32) + await engine._receipt("b" * 32, "a" * 32, {}) + engine.stage_activity.end.assert_awaited_once_with("b" * 32, "a" * 32) + engine.journal.intent.return_value["disposition"] = "completed" + await engine._begin_task("b" * 32, "implement", tmp_path, {}) + assert engine.stage_activity.begin.await_count == 1 + + +class ObservationStore: + schema = "forge_observer_test" + _ownership_pid = 123 + execution_stopping = False + + def __init__(self): + self.execution_ownership_token = object() + self.run = { + "repository_id": uuid4().hex, + "execution_actor": {"authentication_method": "oidc_bearer", "user_id": uuid4().hex}, + } + self.stage = SimpleNamespace(data={"idempotency_key": "a" * 32}, plan_id=uuid4(), attempt=1) + self.session = SimpleNamespace( + scalar=AsyncMock( + side_effect=[ + self.stage, + SimpleNamespace(digest="c" * 64), + SimpleNamespace(id=uuid4(), digest="d" * 64), + datetime.now(UTC), + ] + ) + ) + self.units = 0 + self.failure = False + + async def assert_ownership(self): + if self.failure: + raise RuntimeError("owner lost") + + async def get_run(self, run_id): + return self.run + + @asynccontextmanager + async def _unit(self): + self.units += 1 + yield self.session + + +def observer(): + store = ObservationStore() + records = SimpleNamespace(begin=AsyncMock(return_value=uuid4()), renew=AsyncMock(), end=AsyncMock()) + activity = EngineActivity(store, records, EngineActivitySettings(heartbeat_seconds=0.01, stale_seconds=1)) + return store, records, activity + + +@pytest.mark.asyncio +async def test_blocked_real_engine_stage_is_observed_then_receipt_ends_it(monkeypatch, tmp_path): + store, records, activity = observer() + engine = Engine.__new__(Engine) + engine.store, engine.stage_activity = store, activity + engine.assert_ownership = store.assert_ownership + engine.journal = SimpleNamespace( + ensure_plan=AsyncMock(return_value={"tasks": [{"id": "implement", "attempt": 1, "status": "queued"}]}), + intent=AsyncMock(return_value={"disposition": "execute", "idempotency_key": "a" * 32}), + receipt=AsyncMock(), + ) + monkeypatch.setattr(ProjectAccess, "authorize_execution", AsyncMock()) + entered, release = asyncio.Event(), asyncio.Event() + + async def stage(): + await engine._begin_task("b" * 32, "implement", tmp_path, {}) + entered.set() + await release.wait() + await engine._receipt("b" * 32, "a" * 32, {}) + + task = asyncio.create_task(stage()) + try: + await entered.wait() + assert len(activity.scopes) == 1 and not task.done() + assert next(iter(activity.scopes.values())).task is task + records.renew.return_value = [next(iter(activity.scopes.values())).row_id] + units = store.units + await activity.pulse() + assert store.units == units + 1 + records.renew.assert_awaited_once() + release.set() + await task + assert not activity.scopes + records.end.assert_awaited_once() + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await activity.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["owner", "backend", "expired", "done", "cancelled"]) +async def test_observation_failure_never_replays_or_cancels_business_work(failure): + store, records, activity = observer() + entered = asyncio.Event() + + async def business(): + await activity.begin("b" * 32, "implement", "a" * 32) + entered.set() + await asyncio.Event().wait() + + task = asyncio.create_task(business()) + try: + await entered.wait() + assert activity.scopes + if failure == "owner": + store.failure = True + elif failure == "backend": + records.renew.side_effect = RuntimeError("unavailable") + elif failure == "expired": + records.renew.return_value = [] + else: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await activity.pulse() + assert not activity.scopes + if failure not in {"done", "cancelled"}: + assert not task.done() + records.begin.assert_awaited_once() + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await activity.close() + + +@pytest.mark.asyncio +async def test_heartbeat_uses_empty_context_and_loses_changed_physical_owner(): + store, records, activity = observer() + marker = ContextVar("request_identity", default=None) + marker.set("must_not_leak") + observed = asyncio.Event() + + async def renew(*args): + assert marker.get() is None + observed.set() + return [] + + records.renew.side_effect = renew + try: + await activity.begin("b" * 32, "implement", "a" * 32) + await asyncio.wait_for(observed.wait(), 1) + assert not activity.scopes + finally: + await activity.close() + + +@pytest.mark.asyncio +async def test_end_during_shutdown_keeps_exact_epoch_and_row_identity(): + store, records, activity = observer() + await activity.begin("b" * 32, "implement", "a" * 32) + row = next(iter(activity.scopes.values())).row_id + store.execution_stopping = True + await activity.end("b" * 32) + records.end.assert_awaited_once_with([row], activity.epoch, store.schema) + await activity.close() + + +@pytest.mark.parametrize("settings", [{"heartbeat_seconds": 0}, {"stale_seconds": 5}, {"stale_seconds": float("inf")}]) +def test_staleness_is_a_validated_observation_interval(settings): + with pytest.raises(ValidationError): + EngineActivitySettings(**settings) + + +@pytest.mark.asyncio +async def test_native_queries_keep_exact_correlation_fencing_and_never_revive_expiry(monkeypatch): + captured = [] + session = SimpleNamespace( + scalar=AsyncMock(return_value=None), scalars=AsyncMock(return_value=[]), execute=AsyncMock() + ) + records = ForgeEngineActivityRepository(session=session) + monkeypatch.setattr(records, "_require_session", lambda: session) + for method in (session.scalar, session.scalars, session.execute): + + def capture(statement, *args): + captured.append(str(statement.compile(dialect=postgresql.dialect()))) + return [] + + method.side_effect = capture + await records.renew([uuid4()], "a" * 32, 20, "forge_observer_test") + await records.end([uuid4()], "a" * 32, "forge_observer_test") + renewal, end = captured + assert "forge_engine_activity.expires_at > clock_timestamp()" in renewal + assert "forge_engine_activity.owner_epoch =" in renewal and "forge_engine_activity.owner_epoch =" in end + assert "forge_engine_activity.id IN" in end + assert "FROM pg_catalog.pg_locks, forge_engine_activity" not in renewal + assert "FROM forge_project_memberships, forge_runs" not in renewal + assert "FROM forge_plans \nWHERE forge_plans.run_id = forge_runs.id" in renewal + for required in ( + "forge_tasks.attempt", + "idempotency_key", + "forge_journal.digest", + "api_roles_expires_at", + "backend_start", + "pg_locks", + "scope = jsonb_build_object", + ): + assert required in renewal + + +@pytest.mark.asyncio +async def test_bulk_heartbeat_is_one_unit_for_multiple_current_local_tasks(): + from firefly_forge.engine_activity import LocalStage + + store, records, activity = observer() + task = asyncio.current_task() + assert task is not None + ids = [uuid4(), uuid4()] + for index, row in enumerate(ids): + run_id, key = f"{index + 1:032x}", f"{index + 3:032x}" + activity.scopes[(run_id, key)] = LocalStage(row, run_id, key, task, store.execution_ownership_token) + records.renew.return_value = ids + await activity.pulse() + assert store.units == 1 and len(activity.scopes) == 2 + records.renew.assert_awaited_once_with(ids, activity.epoch, 1, store.schema) + await activity.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("actor", [None, "scm_mapping", "personal_token", "unknown"]) +async def test_unsupported_actor_paths_do_not_claim_live_observations(actor): + store, records, activity = observer() + store.run["execution_actor"] = {"authentication_method": actor} if actor else None + await activity.begin("b" * 32, "implement", "a" * 32) + assert not activity.scopes and activity.poller is None + records.begin.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_slow_observation_times_out_without_cancelling_the_engine_task(): + store, records, activity = observer() + entered = asyncio.Event() + + async def stalled(*args): + entered.set() + await asyncio.Event().wait() + + records.begin.side_effect = stalled + activity.settings = EngineActivitySettings( + heartbeat_seconds=0.1, + stale_seconds=1, + sample_timeout_seconds=0.01, + ) + await activity.begin("b" * 32, "implement", "a" * 32) + assert entered.is_set() and not activity.scopes + assert not asyncio.current_task().cancelling() + + +def cleanup_engine(tmp_path, monkeypatch, observer_failure): + from firefly_forge import checkpoint_mode_runtime + + state = {"entered": False, "closed": False} + run = { + "id": "b" * 32, + "repository_id": "c" * 32, + "status": "preparing", + "repair_budget": 0, + "checkpoint_mode_adoption": {"fixture": True}, + } + + @asynccontextmanager + async def credentials(*args): + state["entered"] = True + try: + yield + finally: + state["closed"] = True + + async def finish(identifier, status, **fields): + run.update(status=status, **fields) + + store = SimpleNamespace( + path=tmp_path / "state", + runtime_roles=None, + configuration=object(), + claim=AsyncMock(return_value=True), + get_run=AsyncMock(side_effect=lambda _: dict(run)), + repository=AsyncMock(side_effect=RuntimeError("business failure")), + finish=AsyncMock(side_effect=finish), + event=AsyncMock(), + release_ownership=AsyncMock(), + ) + activity = SimpleNamespace( + begin=AsyncMock(), end=AsyncMock(side_effect=observer_failure), close=AsyncMock(side_effect=observer_failure) + ) + monkeypatch.setattr(checkpoint_mode_runtime, "mode_execution_scope", credentials) + monkeypatch.setattr(checkpoint_mode_runtime, "execution_settings", lambda *args: {}) + engine = Engine(store, tmp_path / "workspaces", provider_factory=lambda _: None, stage_activity=activity) + return engine, state, run + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [RuntimeError("observer failed"), asyncio.CancelledError("observer cancelled")]) +async def test_failing_observer_cannot_skip_actual_engine_credential_cleanup_or_replace_business_outcome( + tmp_path, monkeypatch, failure +): + engine, state, run = cleanup_engine(tmp_path, monkeypatch, failure) + await engine.execute_run(run["id"]) + assert state == {"entered": True, "closed": True} + assert run["status"] == "failed" and run["error"] == "business failure" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [RuntimeError("observer failed"), asyncio.CancelledError("observer cancelled")]) +async def test_actual_engine_cancellation_still_closes_credentials_when_observer_fails(tmp_path, monkeypatch, failure): + engine, state, run = cleanup_engine(tmp_path, monkeypatch, failure) + blocked = asyncio.Event() + + async def repository(*args): + blocked.set() + await asyncio.Event().wait() + + engine.store.repository.side_effect = repository + task = asyncio.create_task(engine.execute_run(run["id"])) + await asyncio.wait_for(blocked.wait(), 1) + task.cancel("operator requested") + with pytest.raises(asyncio.CancelledError) as cancelled: + await task + assert cancelled.value.args == ("operator requested",) + assert state == {"entered": True, "closed": True} + assert run["status"] == "cancelled" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [RuntimeError("observer failed"), asyncio.CancelledError("observer cancelled")]) +async def test_failing_observer_cannot_skip_drained_engine_owner_release(tmp_path, monkeypatch, failure): + engine, _, _ = cleanup_engine(tmp_path, monkeypatch, failure) + engine.enforce_ownership = True + engine.owner_lock.acquire(timeout=0) + try: + await engine.stop() + engine.store.release_ownership.assert_awaited_once() + assert not engine.owner_lock.is_locked + finally: + if engine.owner_lock.is_locked: + engine.owner_lock.release() + + +@pytest.mark.asyncio +async def test_cancellation_during_observer_shutdown_releases_drained_owner_and_propagates(tmp_path, monkeypatch): + engine, _, _ = cleanup_engine(tmp_path, monkeypatch, None) + engine.enforce_ownership = True + engine.owner_lock.acquire(timeout=0) + entered = asyncio.Event() + + async def blocked_close(): + entered.set() + await asyncio.Event().wait() + + engine.stage_activity.close.side_effect = blocked_close + task = asyncio.create_task(engine.stop()) + try: + await asyncio.wait_for(entered.wait(), 1) + task.cancel("shutdown requested") + with pytest.raises(asyncio.CancelledError) as cancelled: + await task + assert cancelled.value.args == ("shutdown requested",) + engine.store.release_ownership.assert_awaited_once() + assert not engine.owner_lock.is_locked + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + if engine.owner_lock.is_locked: + engine.owner_lock.release() + + +@pytest.mark.asyncio +async def test_undrained_consumer_still_holds_owner_without_calling_observer_close(tmp_path, monkeypatch): + engine, _, _ = cleanup_engine(tmp_path, monkeypatch, RuntimeError("observer failed")) + engine.enforce_ownership = True + engine.owner_lock.acquire(timeout=0) + consumer = asyncio.create_task(asyncio.Event().wait()) + engine.consumer_tasks = lambda: [consumer] + try: + await engine.stop() + assert engine.execution_state == "degraded" and engine.owner_lock.is_locked + engine.store.release_ownership.assert_not_awaited() + engine.stage_activity.close.assert_not_awaited() + finally: + consumer.cancel() + await asyncio.gather(consumer, return_exceptions=True) + engine.owner_lock.release() + RETAINED_EXECUTION_FENCES.discard(engine.owner_lock) diff --git a/tests/test_engine_activity_material_host.py b/tests/test_engine_activity_material_host.py new file mode 100644 index 000000000..ce884d78b --- /dev/null +++ b/tests/test_engine_activity_material_host.py @@ -0,0 +1,23 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Stage-observation implementation belongs to prospective runtime material.""" + +from pathlib import Path + +import pytest + +from firefly_forge.agentic_checkpoint_invocation import runtime_contract +from firefly_forge.source_budget_allowances import admission_contract_digest + + +@pytest.mark.parametrize("contract", [runtime_contract, admission_contract_digest]) +@pytest.mark.parametrize("name", ["engine_activity.py", "engine_activity_persistence.py"]) +def test_stage_observation_material_change_invalidates_its_current_contract(contract, name, monkeypatch): + before = contract() + original = Path.read_bytes + + def changed(path): + value = original(path) + return value + b"\n# controlled observation mutation\n" if path.name == name else value + + monkeypatch.setattr(Path, "read_bytes", changed) + assert contract() != before diff --git a/tests/test_engine_activity_postgres.py b/tests/test_engine_activity_postgres.py new file mode 100644 index 000000000..d1e3d64b3 --- /dev/null +++ b/tests/test_engine_activity_postgres.py @@ -0,0 +1,392 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Native observation custody; only the dedicated PostgreSQL fixture is permitted.""" + +import asyncio +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from sqlalchemy import func, select, update +from sqlalchemy.orm import aliased + +from firefly_forge.access_persistence import ForgeMembershipEntity, ForgeMembershipRepository +from firefly_forge.engine import Engine +from firefly_forge.engine_activity import EngineActivity +from firefly_forge.engine_activity_persistence import ( + ForgeEngineActivityEntity, + ForgeEngineActivityRepository, + current_activity_predicate, +) +from firefly_forge.factory_dashboard_models import DashboardScope +from firefly_forge.factory_dashboard_persistence import visible_runs_statement +from firefly_forge.identity import Principal +from firefly_forge.identity_persistence import ForgeIdentityUserEntity, ForgeIdentityUserRepository +from firefly_forge.journal import Journal +from firefly_forge.persistence import ( + ForgePlanEntity, + ForgePlanRepository, + ForgeRepositoryEntity, + ForgeRunEntity, + ForgeTaskEntity, + ForgeTaskRepository, +) +from firefly_forge.project_persistence import ForgeProjectEntity, ForgeProjectRepository + + +@pytest.fixture +async def stage(store, tmp_path): + user, repo, run, plan = (uuid4() for _ in range(4)) + actor = {"user_id": user.hex, "authentication_method": "local_bootstrap", "global_admin": True} + async with store._unit(): + await ForgeIdentityUserRepository().save( + ForgeIdentityUserEntity(id=user, issuer="local:bootstrap", subject="operator", enabled=True) + ) + await store.repository_records.save(ForgeRepositoryEntity(id=repo, data={"id": repo.hex})) + await store.run_records.save( + ForgeRunEntity( + id=run, + repository_id=repo, + status="developing", + data={ + "id": run.hex, + "repository_id": repo.hex, + "execution_actor": actor, + "agent": "agentic", + "status": "developing", + "brief": "Fixture stage", + }, + ) + ) + await ForgePlanRepository().save( + ForgePlanEntity( + id=plan, + run_id=run, + revision=1, + digest="a" * 64, + data={ + "tasks": [ + {"id": "implement", "kind": "local", "title": "Implement", "instructions": "Fixture only"} + ], + }, + ) + ) + await ForgeTaskRepository().save( + ForgeTaskEntity(run_id=run, plan_id=plan, task_id="implement", status="queued", attempt=1, data={}) + ) + await store.acquire_ownership() + activity = EngineActivity(store, ForgeEngineActivityRepository()) + engine = Engine(store, tmp_path / "workspaces", journal=Journal(store), stage_activity=activity) + engine.enforce_ownership = True + value = SimpleNamespace( + store=store, activity=activity, engine=engine, run=run, user=user, repo=repo, plan=plan, path=tmp_path + ) + yield value + await activity.close() + await store.release_ownership() + + +async def current(c): + async with c.store._unit() as session: + return list( + await session.scalars( + select(ForgeEngineActivityEntity) + .select_from(ForgeEngineActivityEntity) + .join(ForgeRunEntity, ForgeRunEntity.id == ForgeEngineActivityEntity.run_id) + .join(ForgeTaskEntity, ForgeTaskEntity.run_id == ForgeRunEntity.id) + .where(current_activity_predicate(c.store.schema)) + ) + ) + + +async def begin(c): + return await c.engine._begin_task(c.run.hex, "implement", c.path, {}) + + +async def test_native_engine_stage_renewal_end_and_late_epoch_cannot_retire_successor(stage): + c = stage + async with c.store._unit(): + await ForgePlanRepository().save( + ForgePlanEntity( + id=uuid4(), + run_id=c.run, + revision=2, + digest="b" * 64, + data={ + "tasks": [ + {"id": "implement", "kind": "local", "title": "Implement", "instructions": "Fixture only"} + ], + }, + ) + ) + intent = await begin(c) + rows = await current(c) + assert len(rows) == 1 and rows[0].invocation_key == intent["idempotency_key"] + assert rows[0].plan_id == c.plan and rows[0].current_plan_id != c.plan + before = rows[0].observed_at + await c.activity.pulse() + assert (await current(c))[0].observed_at > before + old = next(iter(c.activity.scopes.values())) + successor = EngineActivity(c.store, ForgeEngineActivityRepository()) + try: + await successor.begin(c.run.hex, "implement", intent["idempotency_key"]) + assert (await current(c))[0].owner_epoch == successor.epoch + await c.activity.end(c.run.hex) + assert (await current(c))[0].owner_epoch == successor.epoch + async with c.store._unit(): + await ForgeEngineActivityRepository().end([old.row_id], c.activity.epoch, c.store.schema) + assert len(await current(c)) == 1 + await successor.end(c.run.hex) + assert not await current(c) + finally: + await successor.close() + + +@pytest.mark.parametrize("change", ["actor", "attempt", "scope", "plan", "fence", "expiry"]) +async def test_native_current_predicate_refuses_drift_revocation_and_cold_expiry(stage, change): + c = stage + await begin(c) + assert len(await current(c)) == 1 + async with c.store._unit() as session: + if change == "actor": + await session.execute( + update(ForgeIdentityUserEntity).where(ForgeIdentityUserEntity.id == c.user).values(enabled=False) + ) + elif change == "attempt": + await session.execute(update(ForgeTaskEntity).where(ForgeTaskEntity.run_id == c.run).values(attempt=2)) + elif change == "scope": + row = await c.store.run_records.find_by_id(c.run) + row.data = {**row.data, "policy_snapshot": {"digest": "b" * 64}} + await c.store.run_records.save(row) + elif change == "plan": + await ForgePlanRepository().save( + ForgePlanEntity(id=uuid4(), run_id=c.run, revision=2, digest="b" * 64, data={"tasks": []}) + ) + elif change == "expiry": + await session.execute(update(ForgeEngineActivityEntity).values(expires_at=func.clock_timestamp())) + if change == "fence": + await c.store.release_ownership() + assert not await current(c) + await c.activity.pulse() + assert not c.activity.scopes + assert not await current(c) + # Constructing a cold observer never revives a row or adopts an old local task. + cold = EngineActivity(c.store, ForgeEngineActivityRepository()) + assert not cold.scopes + await cold.close() + + +async def test_native_cancelled_blocked_engine_task_ends_observation_without_advancing_journal(stage): + c = stage + entered = asyncio.Event() + + async def work(): + await begin(c) + entered.set() + try: + await asyncio.Event().wait() + finally: + await c.activity.end(c.run.hex) + + task = asyncio.create_task(work()) + try: + await entered.wait() + assert len(await current(c)) == 1 + prior = await c.engine.journal.entries(c.run.hex) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + assert not await current(c) + assert await c.engine.journal.entries(c.run.hex) == prior + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +async def test_native_observation_composes_with_reader_scope_and_current_membership_revocation(stage): + c = stage + project, foreign = uuid4(), uuid4() + async with c.store._unit(): + await ForgeProjectRepository().save( + ForgeProjectEntity(id=project, data={"definition": {"name": "Observed product", "components": []}}) + ) + user = await ForgeIdentityUserRepository().find_by_id(c.user) + user.issuer, user.subject = "https://id.example", "owner" + await ForgeIdentityUserRepository().save(user) + await ForgeIdentityUserRepository().save( + ForgeIdentityUserEntity(id=foreign, issuer="https://id.example", subject="foreign", enabled=True) + ) + await ForgeMembershipRepository().save( + ForgeMembershipEntity(project_id=project, user_id=c.user, active=True, role="developer") + ) + row = await c.store.run_records.find_by_id(c.run) + row.data = { + **row.data, + "project_id": project.hex, + "execution_actor": {"user_id": c.user.hex, "authentication_method": "oidc_bearer"}, + } + await c.store.run_records.save(row) + await begin(c) + assert len(await current(c)) == 1 + stranger = Principal(foreign, "https://id.example", "foreign") + scoped = visible_runs_statement( + DashboardScope(kind="run", run_id=c.run.hex), administrator=False, actor=stranger + ).subquery() + run = aliased(ForgeRunEntity, scoped) + async with c.store._unit() as session: + visible = await session.scalar( + select(ForgeEngineActivityEntity.id) + .join(run, run.id == ForgeEngineActivityEntity.run_id) + .join(ForgeTaskEntity, ForgeTaskEntity.run_id == run.id) + .where(current_activity_predicate(c.store.schema, run=run)) + ) + assert visible is None + await session.execute( + update(ForgeMembershipEntity).where(ForgeMembershipEntity.user_id == c.user).values(active=False) + ) + assert not await current(c) + await c.activity.pulse() + assert not c.activity.scopes + + +async def test_native_dashboard_observes_only_current_engine_stage_then_receipt_ends_it(stage): + from firefly_forge.factory_dashboard import FactoryDashboard + from firefly_forge.factory_dashboard_models import DashboardPageInput + from firefly_forge.factory_dashboard_persistence import FactoryDashboardRepository + from firefly_forge.migrations import SCHEMA_REVISION + from firefly_forge.project_access import ProjectAccess + + c = stage + assert SCHEMA_REVISION == 33 + await begin(c) + actor = Principal(c.user, "local:bootstrap", "operator", authentication_method="local_bootstrap") + dashboard = FactoryDashboard(c.store, ProjectAccess(c.store), FactoryDashboardRepository()) + scope = DashboardScope(kind="run", run_id=c.run.hex) + value = await dashboard.observe(actor, scope, DashboardPageInput()) + observed = value["stages"][1] + assert observed["active"] == 1 and observed["unknown"] == 0 + assert observed["observed_until"] > value["observed_at"] + assert "worker_liveness" in value["unavailable_metrics"] + key = next(iter(c.activity.scopes.values())).key + await c.engine._receipt(c.run.hex, key, {"fixture": "known local stage completion"}) + after = await dashboard.observe(actor, scope, DashboardPageInput()) + assert after["stages"][1]["active"] == 0 and after["stages"][1]["done"] == 1 + assert after["stages"][1]["observed_until"] is None + + +async def test_native_pat_reader_observes_engine_stage_without_execution_or_foreign_scope(stage): + from datetime import UTC, datetime, timedelta + + from firefly_forge.factory_dashboard import FactoryDashboard + from firefly_forge.factory_dashboard_models import DashboardPageInput + from firefly_forge.factory_dashboard_persistence import FactoryDashboardRepository + from firefly_forge.personal_token_models import PersonalTokenInput + from firefly_forge.personal_tokens import PersonalTokens + from firefly_forge.project_access import AccessDenied, ProjectAccess + + c = stage + project, foreign = uuid4(), uuid4() + async with c.store._unit(): + for value in (project, foreign): + await ForgeProjectRepository().save( + ForgeProjectEntity(id=value, data={"definition": {"name": "Observed project", "components": []}}) + ) + user = await ForgeIdentityUserRepository().find_by_id(c.user) + user.issuer, user.subject = "https://id.example", "observer" + user.api_roles_audience = "forge-api" + user.api_roles_issued_at = datetime.now(UTC).timestamp() + user.api_roles_expires_at = datetime.now(UTC) + timedelta(hours=1) + await ForgeMembershipRepository().save( + ForgeMembershipEntity(project_id=project, user_id=c.user, active=True, role="developer") + ) + row = await c.store.run_records.find_by_id(c.run) + row.data = { + **row.data, + "project_id": project.hex, + "agent": "agentic", + "sandbox_snapshot": {"mode": "docker"}, + "execution_actor": { + "user_id": c.user.hex, + "authentication_method": "oidc_bearer", + "api_audience": "forge-api", + }, + } + owner = Principal(c.user, user.issuer, user.subject, authentication_method="oidc_session", api_audience="forge-api") + tokens = PersonalTokens(c.store) + issued = await tokens.create( + owner, + PersonalTokenInput( + request_id=uuid4().hex, + name="Stage observations", + project_ids=[project.hex], + expires_at=datetime.now(UTC) + timedelta(hours=1), + ), + ) + assert issued.secret is not None + reader = (await tokens.resolve(issued.secret.get_secret_value())).principal + await begin(c) + dashboard = FactoryDashboard(c.store, ProjectAccess(c.store, personal_tokens=tokens), FactoryDashboardRepository()) + selected = DashboardScope(kind="project", project_id=project.hex) + value = await dashboard.observe(reader, selected, DashboardPageInput()) + assert value["stages"][1]["active"] == 1 and value["standalone_available"] is False + with pytest.raises(AccessDenied): + await dashboard.observe(reader, DashboardScope(kind="project", project_id=foreign.hex), DashboardPageInput()) + with pytest.raises(AccessDenied): + await dashboard.access.authorize_actor(ProjectAccess.actor_snapshot(reader), project.hex, "execute") + async with c.store._unit(): + membership = (await ForgeMembershipRepository().for_user(c.user))[0] + membership.active = False + with pytest.raises(AccessDenied): + await dashboard.observe(reader, selected, DashboardPageInput()) + + +async def test_native_schema32_upgrade_adds_only_observation_table_and_preserves_all_retained_rows( + store_factory, tmp_path, monkeypatch +): + import hashlib + import importlib.util + from pathlib import Path + + from sqlalchemy import text + from test_analytics_upgrade_native import retained_rows + + import firefly_forge.store as store_module + from firefly_forge.migrations import SCHEMA_REVISION, migrate + + source = Path(__file__).parent / "fixtures/engine_activity/schema32_migrations.py" + assert ( + hashlib.sha256(source.read_bytes()).hexdigest() + == "bfbb0d2a016c9e2563e4aeffacfb004b2b1252c604d07ee7040b1f206f6bf9ea" + ) + spec = importlib.util.spec_from_file_location("owned_engine_activity_schema32", source) + assert spec is not None and spec.loader is not None + legacy = importlib.util.module_from_spec(spec) + spec.loader.exec_module(legacy) + assert legacy.SCHEMA_REVISION == 32 and SCHEMA_REVISION == 33 + monkeypatch.setattr(store_module, "migrate", legacy.migrate) + store = await store_factory(tmp_path / "schema32/factory.db") + user, repo, run = (uuid4() for _ in range(3)) + async with store._unit(): + await ForgeIdentityUserRepository().save( + ForgeIdentityUserEntity(id=user, issuer="https://id.example", subject="preserved-owner", enabled=True) + ) + await store.repository_records.save(ForgeRepositoryEntity(id=repo, data={"id": repo.hex, "name": "Preserved"})) + await store.run_records.save( + ForgeRunEntity( + id=run, + repository_id=repo, + status="paused", + data={"id": run.hex, "attempt": 3, "spent": 2, "status": "paused", "retained": ["prior receipt"]}, + ) + ) + before = await retained_rows(store) + assert "forge_engine_activity" not in before + monkeypatch.setattr(store_module, "migrate", migrate) + async with store._unit() as session: + await migrate(session, store.schema) + await migrate(session, store.schema) + assert list(await session.scalars(text("SELECT revision FROM forge_schema_migrations ORDER BY revision"))) == [ + 32, + 33, + ] + assert await session.scalar(text("SELECT count(*) FROM forge_engine_activity")) == 0 + assert await retained_rows(store, tuple(before)) == before + assert set(await retained_rows(store)) - set(before) == {"forge_engine_activity"} diff --git a/tests/test_factory_dashboard_service_host.py b/tests/test_factory_dashboard_service_host.py index 9c49bfe67..ee3cc7286 100644 --- a/tests/test_factory_dashboard_service_host.py +++ b/tests/test_factory_dashboard_service_host.py @@ -347,3 +347,41 @@ async def revoke_after_permission(*args): await dashboard.service.observe( dashboard.actor, DashboardScope(kind="project", project_id="a" * 32), DashboardPageInput() ) + + +async def test_engine_stage_activity_keeps_earliest_expiry_and_rejects_unbounded_positive(dashboard): + from datetime import timedelta + + original = dashboard.records.snapshot + now = datetime(2026, 10, 30, tzinfo=UTC) + + async def snapshot(*args, **kwargs): + value = await original(*args, **kwargs) + value["stages"] = [ + { + "stage": "implement", + "state": "active", + "count": 2, + "recorded_running": 2, + "observed_until": now + timedelta(seconds=20), + }, + { + "stage": "implement", + "state": "active", + "count": 3, + "recorded_running": 3, + "observed_until": now + timedelta(seconds=10), + }, + {"stage": "verify", "state": "active", "count": 1, "recorded_running": 1, "observed_until": None}, + ] + return value + + dashboard.records.snapshot = snapshot + value = await dashboard.service.observe(dashboard.actor, DashboardScope(), DashboardPageInput()) + implementation, verification = value["stages"][1:3] + assert implementation["active"] == 5 and implementation["unknown"] == 0 + assert implementation["observed_until"] == (now + timedelta(seconds=10)).isoformat() + assert verification["active"] == 0 and verification["unknown"] == 1 + assert verification["observed_until"] is None + assert "worker_liveness" in value["unavailable_metrics"] + assert dashboard.records.calls[0][3]["schema"] == dashboard.service.store.schema diff --git a/tests/test_original_selection_readmission_host.py b/tests/test_original_selection_readmission_host.py index 8ea167fd6..830db6a2d 100644 --- a/tests/test_original_selection_readmission_host.py +++ b/tests/test_original_selection_readmission_host.py @@ -193,6 +193,7 @@ async def test_same_attempt_is_claimed_once_and_effective_connection_is_original await c.selection_service.request(c.run_id, uuid4().hex, readmission_request(candidate)) prepared = await prepare_current_mode(c.journal, c.run_id) engine = object.__new__(Engine) + engine.stage_activity = None engine.store, engine.journal, engine.assert_ownership = c.store, c.journal, c.store.assert_ownership with mode_adoption_scope(prepared): settings = execution_settings(c.store.configuration, c.store.run.data) @@ -261,6 +262,7 @@ async def test_later_factory_feedback_adopts_executes_and_finishes_its_own_verif await c.selection_service.request(c.run_id, uuid4().hex, readmission_request(candidate)) prepared = await prepare_current_mode(c.journal, c.run_id) engine = object.__new__(Engine) + engine.stage_activity = None engine.store, engine.journal, engine.assert_ownership = c.store, c.journal, c.store.assert_ownership with mode_adoption_scope(prepared): await engine._begin_task(c.run_id, "implement", c.workspace, {}) @@ -395,6 +397,7 @@ def outside(revision): await c.selection_service.request(c.run_id, uuid4().hex, readmission_request(candidate)) prepared = await prepare_current_mode(c.journal, c.run_id) engine = object.__new__(Engine) + engine.stage_activity = None engine.store, engine.journal, engine.assert_ownership = c.store, c.journal, c.store.assert_ownership with mode_adoption_scope(prepared): result = await engine._begin_task(c.run_id, "implement", c.workspace, {}) diff --git a/tests/test_personal_tokens_postgres.py b/tests/test_personal_tokens_postgres.py index 9e24e3621..e7dddc0e9 100644 --- a/tests/test_personal_tokens_postgres.py +++ b/tests/test_personal_tokens_postgres.py @@ -95,14 +95,14 @@ async def stored_document(store): async def test_revision31_upgrade_keeps_legacy_data_and_adds_native_jsonb_timezone_and_constraints(pg_tokens): c = pg_tokens - assert c.store.schema.startswith("forge_test_") and SCHEMA_REVISION == 32 + assert c.store.schema.startswith("forge_test_") async with c.store._unit() as session: before = await session.scalar(text("SELECT jsonb_agg(to_jsonb(u) ORDER BY id) FROM forge_identity_users u")) project_before = await session.scalar(text("SELECT jsonb_agg(to_jsonb(p) ORDER BY id) FROM forge_projects p")) # These two new tables are empty and belong solely to this disposable fixture schema. await session.execute(text("DROP TABLE forge_personal_token_events")) await session.execute(text("DROP TABLE forge_personal_tokens")) - await session.execute(text("DELETE FROM forge_schema_migrations WHERE revision=32")) + await session.execute(text("DELETE FROM forge_schema_migrations WHERE revision>31")) await session.execute(text("INSERT INTO forge_schema_migrations(revision) VALUES(31) ON CONFLICT DO NOTHING")) async with c.store._unit() as session: await migrate(session, c.store.schema) @@ -115,7 +115,7 @@ async def test_revision31_upgrade_keeps_legacy_data_and_adds_native_jsonb_timezo await session.scalar(text("SELECT jsonb_agg(to_jsonb(p) ORDER BY id) FROM forge_projects p")) == project_before ) - assert await session.scalar(text("SELECT count(*) FROM forge_schema_migrations WHERE revision=32")) == 1 + assert await session.scalar(text("SELECT max(revision) FROM forge_schema_migrations")) == SCHEMA_REVISION columns = dict( ( await session.execute( diff --git a/tests/test_recovery_generation_host.py b/tests/test_recovery_generation_host.py index a266e99e5..3d8bd3868 100644 --- a/tests/test_recovery_generation_host.py +++ b/tests/test_recovery_generation_host.py @@ -35,6 +35,7 @@ async def pending(): await c.service.request(c.run_id, uuid4().hex, request(candidate)) prepared = await prepare_current_mode(c.journal, c.run_id) engine = object.__new__(Engine) + engine.stage_activity = None engine.store, engine.journal, engine.assert_ownership = c.store, c.journal, c.store.assert_ownership with mode_adoption_scope(prepared): begun = await engine._begin_task(c.run_id, "implement", c.workspace, {}) diff --git a/tests/test_run_creation_upgrade_native.py b/tests/test_run_creation_upgrade_native.py index 7b51de39a..f0bfb1780 100644 --- a/tests/test_run_creation_upgrade_native.py +++ b/tests/test_run_creation_upgrade_native.py @@ -45,7 +45,7 @@ async def test_populated_actual_schema26_upgrade_preserves_execution_and_receipt assert spec is not None and spec.loader is not None legacy = importlib.util.module_from_spec(spec) spec.loader.exec_module(legacy) - assert legacy.SCHEMA_REVISION == 26 and SCHEMA_REVISION == 32 + assert legacy.SCHEMA_REVISION == 26 monkeypatch.setattr(store_module, "migrate", legacy.migrate) store = await store_factory(tmp_path / "schema26" / "factory.db") async with store._unit() as session: