diff --git a/.github/workflows/backend-unit-tests.yml b/.github/workflows/backend-unit-tests.yml index 577bc8cb88..7c1a564693 100644 --- a/.github/workflows/backend-unit-tests.yml +++ b/.github/workflows/backend-unit-tests.yml @@ -112,5 +112,6 @@ jobs: # tests (issue #4120), which read DEDUPE_TEST_POSTGRES_URL. Without this # mapping those tests silently skip and the headline capability is never # exercised in CI. + DEERFLOW_TEST_WORKFLOW_POSTGRES: ${{ env.TEST_POSTGRES_URI }} DEDUPE_TEST_POSTGRES_URL: ${{ env.TEST_POSTGRES_URI }} run: make test-shard SPLITS=4 GROUP=${{ matrix.shard }} diff --git a/AGENTS.md b/AGENTS.md index 84c68ce1f9..4808701c68 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,11 +18,11 @@ memory, subagents and extensible tools. IM channels connect through the Gateway. A single `make dev` / Docker stack runs four cooperating services: -| Service | Port | Role | -| --------------- | ------ | ------------------------------------------------------------------- | -| **Nginx** | `2026` | Unified reverse-proxy entry point — open this in the browser | -| **Gateway API** | `8001` | FastAPI REST API + embedded LangGraph-compatible agent runtime | -| **Frontend** | `3000` | Next.js web interface | +| Service | Port | Role | +| --- | --- | --- | +| **Nginx** | `2026` | Unified reverse-proxy entry point — open this in the browser | +| **Gateway API** | `8001` | FastAPI REST API + embedded LangGraph-compatible agent runtime | +| **Frontend** | `3000` | Next.js web interface | | **Provisioner** | `8002` | Optional — only when sandbox is configured for provisioner/K8s mode | Nginx is the single public entry: it proxies `/api/*` to the Gateway, rewriting @@ -48,25 +48,25 @@ Require explicit bind addresses for new published ports; ``` deer-flow/ -├── Makefile # Root orchestration: drives the full stack (dev/start/stop, docker, setup) -├── config.example.yaml # Template → copy to config.yaml (gitignored) at repo root +├── Makefile # Root orchestration: drives the full stack (dev/start/stop, docker, setup) +├── config.example.yaml # Template → copy to config.yaml (gitignored) at repo root ├── extensions_config.example.json # Template → copy to extensions_config.json (gitignored): MCP servers + skills -├── backend/ # Python backend — see backend/AGENTS.md -│ ├── Makefile # Per-module backend commands (dev, gateway, test, lint, migrate-rev) -│ ├── extensions/sources/ # Deployable snapshots of locally installed Python extensions -│ ├── packages/extension-api/ # deerflow-extension-api package (import: deerflow_extension_api.*) — public extension contract -│ ├── packages/harness/ # deerflow-harness package (import: deerflow.*) — agent framework -│ └── app/ # FastAPI Gateway + IM channels (import: app.*) -├── frontend/ # Next.js frontend (pnpm) — see frontend/AGENTS.md -├── docker/ # docker-compose files, nginx config, provisioner -├── skills/ # Agent skills: public/ (committed), custom/ (gitignored) -│ # Managed integration skill packs are global at .deer-flow/integrations/skills/{provider}/ -│ # Integration credentials and enabled state remain per-user -├── contracts/ # Cross-component JSON contracts (e.g. subagent status, skill review) -├── examples/ # Extension examples: deerflow-extension-{example,bookmarks,agent-teams} -├── scripts/ # Root orchestration scripts invoked by the Makefile (check, configure, doctor, support_bundle, serve, nginx, docker, deploy, setup_wizard) -├── tests/ # Root-level tests (currently tests/skills/ — public skill tests) -└── docs/ # Cross-cutting docs, plans, and design notes +├── backend/ # Python backend — see backend/AGENTS.md +│ ├── Makefile # Per-module backend commands (dev, gateway, test, lint, migrate-rev) +│ ├── extensions/sources/ # Deployable snapshots of locally installed Python extensions +│ ├── packages/extension-api/ # deerflow-extension-api package (import: deerflow_extension_api.*) — public extension contract +│ ├── packages/harness/ # deerflow-harness package (import: deerflow.*) — agent framework +│ └── app/ # FastAPI Gateway + IM channels (import: app.*) +├── frontend/ # Next.js frontend (pnpm) — see frontend/AGENTS.md +├── docker/ # docker-compose files, nginx config, provisioner +├── skills/ # Agent skills: public/ (committed), custom/ (gitignored) +│ # Managed integration skill packs are global at .deer-flow/integrations/skills/{provider}/ +│ # Integration credentials and enabled state remain per-user +├── contracts/ # Cross-component JSON contracts (e.g. subagent status, skill review) +├── examples/ # Extension examples: deerflow-extension-{example,bookmarks,agent-teams} +├── scripts/ # Root orchestration scripts invoked by the Makefile (check, configure, doctor, support_bundle, serve, nginx, docker, deploy, setup_wizard) +├── tests/ # Root-level tests (currently tests/skills/ — public skill tests) +└── docs/ # Cross-cutting docs, plans, and design notes ``` Third-party extensions load from operator-controlled `config.yaml -> plugins:`. diff --git a/README.md b/README.md index 9e9a801f6e..b6c818301d 100644 --- a/README.md +++ b/README.md @@ -3775,7 +3775,7 @@ binds the exact retained output or captured file version (SHA-256); preview and downstream reads use the same immutable bytes. Each child retains host-generated assembly evidence for its resolved instructions, model, tools and Skills. Cancelling a recoverable workflow continues child cleanup even if the requesting browser disconnects. -After a Gateway restart, sign in and select **Recover workflow** to recover a compatible run using retained results and exact-version approvals. Uncertain child admissions require reconciliation; they are never automatically retried. Recovery requires unchanged runtime/configuration and durable Gateway storage. Multi-Gateway failover remains Phase 4; there is no hard monetary budget. Workflow deadlines and child-observation timeouts are +After a Gateway restart, sign in and select **Recover workflow** to recover a compatible run using retained results and exact-version approvals. Uncertain child admissions require reconciliation; they are never automatically retried. Recovery requires unchanged runtime/configuration and durable Gateway storage. The optional PostgreSQL mode supports cross-Gateway controls and session-authorized takeover after lease expiry, with fenced child admission and shared limits; see the extension deployment contract. Expired cancellations can be taken over for cleanup even at workflow admission limits. Definitive pre-admission permission/configuration rejection releases child capacity; uncertain admissions retain their reservations. SQLite remains single-Gateway. The [deployed acceptance and delivery boundary](examples/deerflow-extension-workflows/README.md#deployed-acceptance-and-delivery-boundary) documents real two-Gateway fault tests, session-authorized takeover, and the still-open reconciliation API/UI. There is no hard monetary budget. Workflow deadlines and child-observation timeouts are reported separately from uncertain admission. Accepted launches retain their request identity across inspection refresh failures, with a separate inspection retry action. A runnable example, installation steps and QuickJS source-build prerequisites for other Python/CPU combinations are included. diff --git a/backend/app/gateway/extension_agent_runs.py b/backend/app/gateway/extension_agent_runs.py index 5239ae678d..cc5d1919f0 100644 --- a/backend/app/gateway/extension_agent_runs.py +++ b/backend/app/gateway/extension_agent_runs.py @@ -12,7 +12,7 @@ from collections.abc import Awaitable, Callable, Mapping from contextlib import asynccontextmanager from typing import Any -from deerflow_extension_api.agent_runs import AgentRun, AgentRunError +from deerflow_extension_api.agent_runs import AgentRun, AgentRunError, ExecutionFence from fastapi import HTTPException, Request from pydantic import TypeAdapter, ValidationError @@ -25,6 +25,12 @@ from deerflow.utils.thread_id import ThreadId _THREAD_ID = TypeAdapter(ThreadId) +def _configuration_json_default(value): + if isinstance(value, (set, frozenset)): + return sorted(value, key=lambda item: json.dumps(item, sort_keys=True, default=_configuration_json_default)) + return str(value) + + def _json_object(value: Mapping[str, Any] | None) -> dict[str, Any] | None: if value is None: return None @@ -170,9 +176,12 @@ class _BoundAgentRuns: if config is None: agents[name] = None continue - agents[name] = [config.model_dump(mode="json"), load_agent_soul(name, user_id=self._user_id)] - value = [get_app_config().model_dump(mode="json"), get_extensions_config().model_dump(mode="json"), names, agents] - raw = json.dumps(value, sort_keys=True, ensure_ascii=True, default=str).encode() + agents[name] = [config.model_dump(mode="python"), load_agent_soul(name, user_id=self._user_id)] + value = [get_app_config().model_dump(mode="python"), get_extensions_config().model_dump(mode="python"), names, agents] + # JSON-mode Pydantic dumps turn sets into arbitrarily ordered lists. + # Keep set identity until encoding so independent hash seeds agree, + # while ordered model/tool lists still detect precedence changes. + raw = json.dumps(value, sort_keys=True, ensure_ascii=True, default=_configuration_json_default).encode() key = self._host.revision_key if durable: from app.gateway.auth.config import get_auth_config @@ -189,7 +198,7 @@ class _BoundAgentRuns: response = await create_thread(ThreadCreateRequest(assistant_id=assistant_id, thread_id=thread_id, metadata=_json_object(metadata) or {}), request) return response.thread_id - async def _start(self, *, thread_id, input=None, resume=None, context=None, idempotency_key=None): + async def _start(self, *, thread_id, input=None, resume=None, context=None, idempotency_key=None, execution_fence=None): from app.gateway.routers.thread_runs import create_run from app.gateway.run_models import RunCreateRequest from app.gateway.services import resolve_thread_assistant_id @@ -203,13 +212,21 @@ class _BoundAgentRuns: if run_context is not None and any(key in run_context for key in ("agent_name", "is_bootstrap")): raise AgentRunError(422, "Agent identity is bound to the thread") async with self._request("runs:create", thread_id) as request: + if execution_fence is not None: + if not isinstance(execution_fence, ExecutionFence) or self._plugin_namespace is None or execution_fence.namespace != self._plugin_namespace: + raise AgentRunError(422, "Execution fence must belong to this plugin") + from app.gateway.deps import get_run_manager, get_stream_bridge + + if not get_run_manager(request).heartbeat_enabled or getattr(get_stream_bridge(request), "supports_cross_process", False) is not True: + raise AgentRunError(503, "Fenced Agent admission requires run heartbeats and a cross-process stream bridge") + request.state.extension_execution_fence = execution_fence assistant_id = await resolve_thread_assistant_id(request, thread_id, fail_closed=True) body = RunCreateRequest(assistant_id=assistant_id, input=run_input, command=command, context=run_context, on_disconnect="continue") response = await create_run(thread_id, body, request, idempotency_key=f"extension:{self._plugin_namespace or ''}:{idempotency_key}" if idempotency_key is not None else None) return _view(response) - async def start(self, *, thread_id, input, context=None, idempotency_key=None) -> AgentRun: - return await self._start(thread_id=thread_id, input=input, context=context, idempotency_key=idempotency_key) + async def start(self, *, thread_id, input, context=None, idempotency_key=None, execution_fence=None) -> AgentRun: + return await self._start(thread_id=thread_id, input=input, context=context, idempotency_key=idempotency_key, execution_fence=execution_fence) async def resume(self, *, thread_id, resume, idempotency_key=None) -> AgentRun: if resume is None: @@ -228,11 +245,26 @@ class _BoundAgentRuns: async with self._request("threads:read", thread_id) as request: return (await get_thread_state(thread_id, request)).model_dump(mode="json") - async def cancel(self, *, thread_id, run_id) -> None: + async def cancel(self, *, thread_id, run_id, execution_fence=None) -> None: from app.gateway.routers.thread_runs import cancel_run async with self._request("runs:cancel", thread_id) as request: - await cancel_run(thread_id, run_id, request, wait=False, action="interrupt") + if execution_fence is not None: + from app.gateway.deps import get_run_manager + from deerflow.runtime.runs.manager import ConflictError, RunIdempotencyUnsupported + + if not isinstance(execution_fence, ExecutionFence) or self._plugin_namespace is None or execution_fence.namespace != self._plugin_namespace: + raise AgentRunError(422, "Execution fence must belong to this plugin") + manager = get_run_manager(request) + run = await manager.get(run_id, user_id=self._user_id) + if run is None or run.thread_id != thread_id: + raise AgentRunError(404, "Run unavailable") + try: + await manager.request_fenced_cancel(run_id, execution_fence=execution_fence, user_id=self._user_id) + except (ConflictError, RunIdempotencyUnsupported) as exc: + raise AgentRunError(409, str(exc)) from exc + else: + await cancel_run(thread_id, run_id, request, wait=False, action="interrupt") async def wait(self, *, thread_id, run_id, timeout=60) -> AgentRun: if not isinstance(timeout, (int, float)) or not math.isfinite(timeout) or not 0 < timeout <= 3600: diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 0b2cb6d31c..808155777b 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -2550,6 +2550,7 @@ async def start_run( # Bootstrap and unresolved legacy identities stay read-only. evidence_agent_id=(None if scope_runtime_config.get("is_bootstrap") else (_DEFAULT_ASSISTANT_ID if scope_assistant_id == _DEFAULT_ASSISTANT_ID else getattr(agent_config, "name", None))), run_id=admission_run_id, + **({"execution_fence": request.state.extension_execution_fence} if getattr(request.state, "extension_execution_fence", None) is not None else {}), ) except RunIdempotencyUnsupported as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc diff --git a/backend/packages/extension-api/deerflow_extension_api/__init__.py b/backend/packages/extension-api/deerflow_extension_api/__init__.py index fb5d1f43a2..d8d5c56f7b 100644 --- a/backend/packages/extension-api/deerflow_extension_api/__init__.py +++ b/backend/packages/extension-api/deerflow_extension_api/__init__.py @@ -7,7 +7,7 @@ extensions can therefore be released independently of the host. from __future__ import annotations -from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY, AGENT_RUNS_RESOLVER_KEY, AgentRun, AgentRunError, AgentRuns, require_agent_runs, resolve_agent_runs +from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY, AGENT_RUNS_RESOLVER_KEY, AgentRun, AgentRunError, AgentRuns, ExecutionFence, require_agent_runs, resolve_agent_runs from deerflow_extension_api.assembly import ( AgentAssemblyDescriptor, AgentAssemblyObserver, @@ -102,7 +102,7 @@ from deerflow_extension_api.state import ExtensionData #: Contract version. Before 1.0, minors may break and patches are additive. #: From 1.0 on, bump the major for breaking changes. -API_VERSION = "0.2.8" +API_VERSION = "0.2.9" __all__ = [ "AGENT_RUNS_CONTEXT_KEY", @@ -110,6 +110,7 @@ __all__ = [ "AgentRun", "AgentRunError", "AgentRuns", + "ExecutionFence", "require_agent_runs", "resolve_agent_runs", "ModelInvocationError", diff --git a/backend/packages/extension-api/deerflow_extension_api/agent_runs.py b/backend/packages/extension-api/deerflow_extension_api/agent_runs.py index 22af298e02..ced80c0659 100644 --- a/backend/packages/extension-api/deerflow_extension_api/agent_runs.py +++ b/backend/packages/extension-api/deerflow_extension_api/agent_runs.py @@ -18,6 +18,38 @@ class AgentRunError(Exception): self.status_code = status_code +@dataclass(frozen=True) +class ExecutionFence: + """Opaque execution generation, checked atomically with durable run admission. + + The lease must exist in the host database and belong to this capability's + user and plugin namespace. Unsupported stores fail closed. This fences new + admission/cancellation, not arbitrary external effects from an admitted Agent. + + SQL host schema contract: extension_execution_leases has the composite key + (namespace, resource_id), plus user_id, holder_id, epoch (BIGINT), and + expires_at (database-clock Unix seconds). Trusted operator-installed extensions + coordinate these rows through the shared host database. Advance epoch on each + takeover, retain tombstones, and serialize acquisition against row locks. + The host locks that row and checks user/holder/epoch/expiry inside admission + and cancellation transactions. A separate database or preflight check cannot + provide this contract. Fenced calls require heartbeat-enabled run ownership; + start additionally requires a cross-process stream bridge. + """ + + namespace: str + resource_id: str + holder_id: str + epoch: int + + def __post_init__(self): + for value, maximum in ((self.namespace, 96), (self.resource_id, 128), (self.holder_id, 128)): + if not isinstance(value, str) or not value or len(value) > maximum: + raise ValueError("Invalid execution fence identity") + if type(self.epoch) is not int or not 1 <= self.epoch < 2**63: + raise ValueError("Invalid execution fence epoch") + + @dataclass(frozen=True) class AgentRun: thread_id: str @@ -76,11 +108,12 @@ class AgentRuns(Protocol): """Create an owned thread, or return the caller's existing thread ID.""" ... - async def start(self, *, thread_id: str, input: Mapping[str, Any], context: Mapping[str, Any] | None = None, idempotency_key: str | None = None) -> AgentRun: + async def start(self, *, thread_id: str, input: Mapping[str, Any], context: Mapping[str, Any] | None = None, idempotency_key: str | None = None, execution_fence: ExecutionFence | None = None) -> AgentRun: """Start or continue the thread's Agent with ordinary run admission. Scoped handles namespace local idempotency keys automatically. An active - run conflicts; start never implicitly cancels it. + run conflicts; start never implicitly cancels it. execution_fence is + host-only admission authority, never copied into Agent input/context. """ ... @@ -107,8 +140,12 @@ class AgentRuns(Protocol): """ ... - async def cancel(self, *, thread_id: str, run_id: str) -> None: - """Request interrupt, retaining checkpoints. Completion may be async.""" + async def cancel(self, *, thread_id: str, run_id: str, execution_fence: ExecutionFence | None = None) -> None: + """Request interrupt, retaining checkpoints. Completion may be async. + + With execution_fence, persist only under the live extension generation; + the existing run owner consumes the request. Unsupported hosts fail closed. + """ ... diff --git a/backend/packages/extension-api/pyproject.toml b/backend/packages/extension-api/pyproject.toml index d9642e6717..399d992635 100644 --- a/backend/packages/extension-api/pyproject.toml +++ b/backend/packages/extension-api/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "deerflow-extension-api" -version = "0.2.8" +version = "0.2.9" description = "Public contracts for DeerFlow extensions" requires-python = ">=3.12" # Keep the contract package import-light and independent from the host. Public diff --git a/backend/packages/harness/deerflow/extensions/AGENTS.md b/backend/packages/harness/deerflow/extensions/AGENTS.md index d134099015..4c2746ff3f 100644 --- a/backend/packages/harness/deerflow/extensions/AGENTS.md +++ b/backend/packages/harness/deerflow/extensions/AGENTS.md @@ -458,6 +458,9 @@ idempotency isolation; request-resolved handles use `for_plugin` explicitly. `configuration_revision` (API 0.2.7) rechecks grants and returns a process-local HMAC of declarative app/extensions and Agent config/SOUL, not effective assembly, memory, referenced files or provider state. Never expose raw secret configuration. +Durable revisions use the shared host signing secret. Preserve Python set identity +until canonical encoding: unordered config sets must hash identically across +process hash seeds, while ordered model/tool lists must retain their order. `configuration_evidence` reads persisted, run-bound actual assembly versions; the worker captures descriptors even without observers and persists before execution. `read_artifact` authorizes bounded output bytes off-loop; extensions retain immutable @@ -473,3 +476,9 @@ plugin boundary, and the lifecycle hook would be a durable core contract with a single consumer. The generator sits behind the narrow `enqueue_summary` interface so a future document-lifecycle hook can lift generation into an extension without touching schema or rendering. + +Extension API 0.2.9 adds `ExecutionFence` to AgentRuns.start/cancel. Accept only the +capability's plugin namespace and recheck the authenticated user as usual. Stamp +fences on internal request state, never HTTP context/config. Forward admission to +the RunStore transaction; require run heartbeats and a shared stream bridge. Fenced +cancel uses the durable cancellation request, consumed by the existing child owner. diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index d80a0790e8..8a296c8657 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -80,3 +80,15 @@ Unsupported owned calls fail closed; single-worker legacy calls remain supported These are scoped guards, not a generation protocol covering every mutation. Contract/lock-wait tests live in `tests/test_terminal_store_contract.py` (Postgres uses `TEST_POSTGRES_URI`). + +## Extension execution fences + +Migration 0047 owns `extension_execution_leases`. Public `ExecutionFence` carries +namespace/resource/holder/epoch, not authority: Gateway binds it to the current +plugin and user. SQL run admission and fenced cancellation take the run change +clock first, then lock the lease row and check the database clock after lock wait, +retaining that lock through commit. Unsupported stores fail closed. Fenced cancel +persists a request for the existing run owner; never fall back to local cancellation +when the extension generation is stale. Extension-owned state/quota/control logic +stays in its package. Tests: `test_extension_execution_fence.py` (PostgreSQL via +`DEERFLOW_TEST_WORKFLOW_POSTGRES`). diff --git a/backend/packages/harness/deerflow/persistence/extension_leases/__init__.py b/backend/packages/harness/deerflow/persistence/extension_leases/__init__.py new file mode 100644 index 0000000000..c9bfeb3a8c --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/extension_leases/__init__.py @@ -0,0 +1 @@ +"""Generic extension execution ownership and atomic admission fencing.""" diff --git a/backend/packages/harness/deerflow/persistence/extension_leases/fence.py b/backend/packages/harness/deerflow/persistence/extension_leases/fence.py new file mode 100644 index 0000000000..59e8f473c8 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/extension_leases/fence.py @@ -0,0 +1,25 @@ +"""Validate under a row lock retained through Agent admission commit.""" + +from sqlalchemy import select, text + +from .model import ExtensionExecutionLeaseRow + + +async def lock_execution_fence(session, fence, user_id): + from deerflow.runtime.runs.manager import ConflictError + + row = ( + await session.execute( + select(ExtensionExecutionLeaseRow) + .where( + ExtensionExecutionLeaseRow.namespace == fence.namespace, + ExtensionExecutionLeaseRow.resource_id == fence.resource_id, + ) + .with_for_update() + ) + ).scalar_one_or_none() + # Read the database clock AFTER any lock wait, never transaction-start now(). + postgres = session.get_bind().dialect.name == "postgresql" + now = await session.scalar(text("SELECT extract(epoch from clock_timestamp())" if postgres else "SELECT (julianday('now') - 2440587.5) * 86400.0")) + if row is None or row.user_id != user_id or row.holder_id != fence.holder_id or row.epoch != fence.epoch or row.expires_at <= float(now): + raise ConflictError("Stale or unavailable extension execution fence") diff --git a/backend/packages/harness/deerflow/persistence/extension_leases/model.py b/backend/packages/harness/deerflow/persistence/extension_leases/model.py new file mode 100644 index 0000000000..fc378a9e39 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/extension_leases/model.py @@ -0,0 +1,17 @@ +"""Lease rows shared by extension coordination and the host admission transaction.""" + +from sqlalchemy import BigInteger, Double, String +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base + + +class ExtensionExecutionLeaseRow(Base): + __tablename__ = "extension_execution_leases" + + namespace: Mapped[str] = mapped_column(String(96), primary_key=True) + resource_id: Mapped[str] = mapped_column(String(128), primary_key=True) + user_id: Mapped[str] = mapped_column(String(64), nullable=False) + holder_id: Mapped[str] = mapped_column(String(128), nullable=False) + epoch: Mapped[int] = mapped_column(BigInteger, nullable=False) + expires_at: Mapped[float] = mapped_column(Double, nullable=False) diff --git a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md index f47401e358..116a7f7d31 100644 --- a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md @@ -34,7 +34,7 @@ The empty-DB path keeps using `create_all` because `Base.metadata` is the only a `0039_user_disabled` → `0040_scheduled_group_notice` → `0041_im_notice_preferences` → `0042_channel_leases` → `0043_runtime_im_notices` → `0044_channel_runtime_configs` → -`0045_completed_run_evidence` → `0046_skill_mutations` → `0047_mcp_oauth_tokens` (current head). The preference +`0045_completed_run_evidence` → `0046_skill_mutations` → `0047_mcp_oauth_tokens` → `0048_extension_execution_leases` (current head). The preference revision adds a separate owner/key table with a cascading users foreign key and does not alter users; the project-documents revision adds a new owner-scoped shelf table, and the MCP lease-token revision adds two nullable token columns to @@ -204,3 +204,6 @@ Tests: `test_migration_0041_im_notice_preferences.py`. - `0045_completed_run_evidence` adds nullable evidence sealing fields and completed-run snapshots after `0044_channel_runtime_configs`. - `0046_skill_mutations` adds durable owner enrollment, asset revisions, proposals, publication operations and scan leases. - `0047_mcp_oauth_tokens` creates the table-guarded `mcp_oauth_tokens` (PK `scope`, `owner_user_id` (`''` for deployment), `server_key`; encrypted token columns, chain and request fingerprints, `version`, refresh lease) with no backfill. The single-head pin lives in `tests/test_migration_0047_mcp_oauth_tokens.py`. + +- `0048_extension_execution_leases` adds generic extension execution fences after + the shipped OAuth revision; test upgrade/downgrade in `test_extension_execution_fence.py`. diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0048_extension_execution_leases.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0048_extension_execution_leases.py new file mode 100644 index 0000000000..bc03c8295d --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0048_extension_execution_leases.py @@ -0,0 +1,30 @@ +"""Generic extension execution fences. + +Revision ID: 0048_extension_execution_leases +Revises: 0047_mcp_oauth_tokens +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0048_extension_execution_leases" +down_revision = "0047_mcp_oauth_tokens" +branch_labels = None +depends_on = None + + +def upgrade(): + if not sa.inspect(op.get_bind()).has_table("extension_execution_leases"): + op.create_table( + "extension_execution_leases", + sa.Column("namespace", sa.String(96), primary_key=True), + sa.Column("resource_id", sa.String(128), primary_key=True), + sa.Column("user_id", sa.String(64), nullable=False), + sa.Column("holder_id", sa.String(128), nullable=False), + sa.Column("epoch", sa.BigInteger(), nullable=False), + sa.Column("expires_at", sa.Double(), nullable=False), + ) + + +def downgrade(): + op.drop_table("extension_execution_leases") diff --git a/backend/packages/harness/deerflow/persistence/models/__init__.py b/backend/packages/harness/deerflow/persistence/models/__init__.py index a9d7e1a818..19ba90a795 100644 --- a/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -24,6 +24,7 @@ from deerflow.persistence.channel_connections.model import ( from deerflow.persistence.channel_leases.model import ChannelLeaseRow from deerflow.persistence.channel_runtime_configs.model import ChannelRuntimeConfigRow from deerflow.persistence.channel_thread_bindings.model import ChannelThreadBindingRow +from deerflow.persistence.extension_leases.model import ExtensionExecutionLeaseRow from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.login_throttle.model import LoginThrottleRow from deerflow.persistence.managed_subagents.model import ManagedSubagentRow @@ -46,6 +47,7 @@ from deerflow.persistence.user.model import UserPreferenceRow, UserRow from deerflow.persistence.webhook_delivery.model import WebhookDeliveryRow __all__ = [ + "ExtensionExecutionLeaseRow", "AgentRow", "ChannelConnectionRow", "ChannelConversationRow", diff --git a/backend/packages/harness/deerflow/persistence/run/sql.py b/backend/packages/harness/deerflow/persistence/run/sql.py index abe37ef6c5..ce229dc7b1 100644 --- a/backend/packages/harness/deerflow/persistence/run/sql.py +++ b/backend/packages/harness/deerflow/persistence/run/sql.py @@ -807,17 +807,22 @@ class RunRepository(RunStore): return LeaseRenewal(renewed=False) return LeaseRenewal(renewed=True, cancel_action=row.cancel_action) - async def request_cancel(self, run_id: str, *, action: str) -> str | None: + async def request_cancel(self, run_id: str, *, action: str, execution_fence=None, user_id=None) -> str | None: """Atomically persist the first cancellation action on an active run.""" if action not in ("interrupt", "rollback"): raise ValueError(f"Unsupported cancellation action: {action}") now = datetime.now(UTC) async with self._sf() as session: change_seq = await self._next_change_seq(session) + if execution_fence is not None: + from deerflow.persistence.extension_leases.fence import lock_execution_fence + + await lock_execution_fence(session, execution_fence, user_id) result = await session.execute( update(RunRow) .where( RunRow.run_id == run_id, + *([RunRow.user_id == user_id] if execution_fence is not None else []), RunRow.status.in_(("pending", "running")), ) .values( @@ -1008,6 +1013,7 @@ class RunRepository(RunStore): evidence_origin: str = "unknown", evidence_agent_id: str | None = None, idempotency_request: dict[str, Any] | None = None, + execution_fence=None, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """Atomically create a run with cross-process thread-uniqueness. @@ -1057,6 +1063,10 @@ class RunRepository(RunStore): # mutator. One position covers this atomic set of changes; run_id # provides deterministic ordering within the position. change_seq = await self._next_change_seq(session) + if execution_fence is not None: + from deerflow.persistence.extension_leases.fence import lock_execution_fence + + await lock_execution_fence(session, execution_fence, resolved_user_id) claimed: list[dict[str, Any]] = [] if idempotency_key is not None: diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index 98d9208884..64dbeeea29 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -1697,6 +1697,17 @@ class RunManager: record.task.cancel() logger.info("Run %s cancellation signalled locally (action=%s)", run_id, action) + async def request_fenced_cancel(self, run_id, *, execution_fence, user_id): + """Persist cancellation under the extension epoch; the run owner consumes it.""" + import inspect + + if not self.heartbeat_enabled or self._store is None or "execution_fence" not in inspect.signature(self._store.request_cancel).parameters: + raise RunIdempotencyUnsupported("The configured RunStore does not support fenced cancellation") + record = await self.get(run_id, user_id=user_id) + if record is None: + raise ConflictError("Run unavailable") + await self._store.request_cancel(run_id, action="interrupt", execution_fence=execution_fence, user_id=user_id) + async def cancel(self, run_id: str, *, action: str = "interrupt") -> CancelOutcome: """Request cancellation of a run. @@ -1918,6 +1929,7 @@ class RunManager: evidence_agent_id: str | None = None, idempotency_request: dict[str, Any] | None = None, run_id: str | None = None, + execution_fence=None, ) -> RunRecord: """Atomically admit a normal agent run for a thread.""" return await self._admit_thread_operation( @@ -1935,6 +1947,7 @@ class RunManager: evidence_agent_id=evidence_agent_id, idempotency_request=idempotency_request, run_id=run_id, + execution_fence=execution_fence, ) async def _close_cancelled_admission(self, record: RunRecord) -> None: @@ -2008,6 +2021,7 @@ class RunManager: evidence_agent_id: str | None = None, idempotency_request: dict[str, Any] | None = None, run_id: str | None = None, + execution_fence=None, ) -> RunRecord: """Atomically check for inflight runs and create a new one. @@ -2038,6 +2052,9 @@ class RunManager: lease_expires_at = self._compute_lease_expires_at() grace_seconds = self._run_ownership_config.grace_seconds if self._run_ownership_config else 10 + if execution_fence is not None and (self._store is None or not _store_accepts_idempotency_parameter(self._store, "execution_fence") or multitask_strategy != "reject"): + raise RunIdempotencyUnsupported("The configured RunStore does not support execution fencing") + interrupted_records: list[RunRecord] = [] record = RunRecord( run_id=run_id, @@ -2062,7 +2079,7 @@ class RunManager: ) async with self._lock: - if idempotency_key is not None: + if idempotency_key is not None and execution_fence is None: for existing in self._runs.values(): if existing.idempotency_key != idempotency_key: continue @@ -2129,6 +2146,8 @@ class RunManager: "created_at": now, "grace_seconds": grace_seconds, } + if execution_fence is not None: + create_kwargs["execution_fence"] = execution_fence if idempotency_key is not None: create_kwargs["idempotency_key"] = idempotency_key if evidence_origin != "unknown": diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index 97947e4c59..b78e34f1fe 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -13,7 +13,7 @@ dependencies = [ # the contract version it implements, extensions declare ranges. A range # here would let pip resolve a newer contract package than this harness # implements, making newer extensions look supported at runtime. - "deerflow-extension-api==0.2.8", + "deerflow-extension-api==0.2.9", "dotenv>=0.9.9", "exa-py>=1.0.0", "httpx>=0.28.0", diff --git a/backend/tests/support/workflow_gateway.py b/backend/tests/support/workflow_gateway.py new file mode 100644 index 0000000000..dd007cd0a5 --- /dev/null +++ b/backend/tests/support/workflow_gateway.py @@ -0,0 +1,120 @@ +"""Test-only Gateway entrypoint: real host, deterministic model and fault barriers. + +Never deployed or imported by production. Barriers only delay real calls; they do +not replace auth, AgentRuns, run admission, checkpointing, Redis or workflow storage. +""" + +import asyncio +import json +import os +from pathlib import Path + +from langchain_core.language_models import BaseChatModel +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from langchain_core.outputs import ChatGeneration, ChatResult + +EVIDENCE = "immutable acceptance evidence\n版本一\n" + + +class AcceptanceModel(BaseChatModel): + @property + def _llm_type(self): + return "workflow-acceptance-synthetic" + + def bind_tools(self, tools, **kwargs): + return self + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + human = next(m for m in reversed(messages) if isinstance(m, HumanMessage)) + if "HOLD_MODEL" in str(human.content): + import time + + while (Path(os.environ["WORKFLOW_FAULT_DIR"]) / "hold_model").exists(): + time.sleep(0.02) + if "CREATE_EVIDENCE" in str(human.content) and not any(isinstance(m, ToolMessage) for m in messages): + answer = AIMessage(content="", tool_calls=[{"name": "write_file", "args": {"path": "/mnt/user-data/outputs/evidence.txt", "content": EVIDENCE}, "id": "evidence-write", "type": "tool_call"}]) + elif "CREATE_EVIDENCE" in str(human.content) and not any(isinstance(m, ToolMessage) and m.name == "present_files" for m in messages): + answer = AIMessage(content="", tool_calls=[{"name": "present_files", "args": {"filepaths": ["/mnt/user-data/outputs/evidence.txt"]}, "id": "evidence-present", "type": "tool_call"}]) + else: + content = str(human.content) + if "--- BEGIN USER INPUT ---\n" in content: + content = content.split("--- BEGIN USER INPUT ---\n", 1)[1].split("\n--- END USER INPUT ---", 1)[0] + answer = AIMessage(content=json.dumps({"echo": content})) + return ChatResult(generations=[ChatGeneration(message=answer)]) + + +async def barrier(boundary, record=None): + root = Path(os.environ["WORKFLOW_FAULT_DIR"]) + + def armed(): + arm = root / "arm.json" + if not arm.exists(): + return False + data = json.loads(arm.read_text()) + if data["boundary"] != boundary: + return False + arm.unlink() + (root / "reached.json").write_text(json.dumps({"boundary": boundary, "run_id": record.get("id") if record else None})) + return True + + if await asyncio.to_thread(armed): + while not await asyncio.to_thread((root / "release").exists): + await asyncio.sleep(0.02) + + +def install_barriers(): + from deerflow_extension_workflows.postgres import PostgresStore + + from app.gateway.extension_agent_runs import _BoundAgentRuns + + original_put = PostgresStore.put + + def put(self, owner, record): + if any(s["status"] == "completed" for s in record["steps"]): + asyncio.run(barrier("result_before", record)) + result = original_put(self, owner, record) + boundary = None + if any(s["status"] == "completed" for s in record["steps"]): + boundary = "result" + if any(c["status"] == "approved" for c in record["checkpoints"]): + boundary = "approval" + if record["status"] == "cancelling": + boundary = "cancelling" + if boundary: + asyncio.run(barrier(boundary, record)) + return result + + PostgresStore.put = put + original_capture = PostgresStore.capture_artifact + + def capture(self, *args, **kwargs): + asyncio.run(barrier("capture")) + try: + return original_capture(self, *args, **kwargs) + except Exception as exc: + (Path(os.environ["WORKFLOW_FAULT_DIR"]) / "capture_error").write_text(type(exc).__name__) + raise + + PostgresStore.capture_artifact = capture + original_start = _BoundAgentRuns.start + + async def start(self, **kwargs): + await barrier("before_admission") + try: + result = await original_start(self, **kwargs) + except Exception as exc: + await asyncio.to_thread((Path(os.environ["WORKFLOW_FAULT_DIR"]) / "admission_error").write_text, type(exc).__name__) + raise + await barrier("admission") + return result + + _BoundAgentRuns.start = start + + +if __name__ == "__main__": + import sys + + import uvicorn + + install_barriers() + uvicorn.run("app.gateway.app:app", host="127.0.0.1", port=int(sys.argv[1]), log_level="warning") diff --git a/backend/tests/test_extension_agent_runs.py b/backend/tests/test_extension_agent_runs.py index db8771a540..230c1b4f03 100644 --- a/backend/tests/test_extension_agent_runs.py +++ b/backend/tests/test_extension_agent_runs.py @@ -466,3 +466,77 @@ async def test_durable_configuration_revision_survives_host_replacement_and_rota load_user.return_value = None with pytest.raises(AgentRunError): await second.configuration_revision(assistant_ids=["lead_agent"], durable=True) + + +@pytest.mark.asyncio +async def test_execution_fence_namespace_and_host_prerequisites(runtime): + from deerflow_extension_api import ExecutionFence + + runs, state, request, _, _ = runtime + runs = runs.for_plugin("test.fenced") + thread_id = await runs.create_thread(thread_id="fenced-thread") + with pytest.raises(AgentRunError, match="belong to this plugin"): + await runs.start(thread_id=thread_id, input={}, execution_fence=ExecutionFence("other", "workflow", "owner", 1)) + with pytest.raises(AgentRunError, match="heartbeats"): + await runs.start(thread_id=thread_id, input={}, execution_fence=ExecutionFence("test.fenced", "workflow", "owner", 1)) + + +@pytest.mark.asyncio +async def test_host_forwards_fence_to_atomic_admission(runtime, monkeypatch, tmp_path): + from deerflow_extension_api import ExecutionFence + from sqlalchemy import insert + from test_run_repository import _cleanup, _make_repo + + from deerflow.config.run_ownership_config import RunOwnershipConfig + from deerflow.persistence.extension_leases.model import ExtensionExecutionLeaseRow + from deerflow.runtime import RunManager + + runs, state, request, _, _ = runtime + repository = await _make_repo(tmp_path) + state.run_store = repository + await state.run_manager.shutdown() + state.run_manager = RunManager(store=repository, run_ownership_config=RunOwnershipConfig(heartbeat_enabled=True)) + # Transport is irrelevant to this admission test; advertise its prerequisite + # without starting Redis. The actual host/manager/store admission path is real. + monkeypatch.setattr(type(state.stream_bridge), "supports_cross_process", True, raising=False) + monkeypatch.setattr(type(state.stream_bridge), "start_producer_lease", AsyncMock(return_value=True), raising=False) + runs = runs.for_plugin("test.fenced") + thread_id = await runs.create_thread(thread_id="fenced-host-thread") + async with repository._sf() as session: + await session.execute(insert(ExtensionExecutionLeaseRow).values(namespace="test.fenced", resource_id="workflow", user_id=str(request.state.user.id), holder_id="new", epoch=2, expires_at=9999999999)) + await session.commit() + try: + with pytest.raises(AgentRunError) as failure: + await runs.start(thread_id=thread_id, input={"messages": [{"role": "user", "content": "must not execute"}]}, execution_fence=ExecutionFence("test.fenced", "workflow", "old", 1)) + assert failure.value.status_code == 409 + assert "execution fence" in str(failure.value) + finally: + await state.run_manager.shutdown() + await _cleanup() + + +@pytest.mark.asyncio +async def test_configuration_revision_canonicalizes_sets_but_preserves_list_order(host, monkeypatch): + class OrderedSet(set): + def __init__(self, values): + super().__init__(values) + self.values = values + + def __iter__(self): + return iter(self.values) + + control, request, _ = host + data = {"tools": ["first", "second"], "exempt_tools": OrderedSet(["task", "present_files"])} + + # Pydantic JSON mode loses the distinction between sets and ordered lists. + def dump(**kwargs): + return {**data, "exempt_tools": list(data["exempt_tools"])} if kwargs.get("mode") == "json" else data + + monkeypatch.setattr("deerflow.config.app_config.get_app_config", lambda: SimpleNamespace(model_dump=dump)) + monkeypatch.setattr("deerflow.config.extensions_config.get_extensions_config", lambda: SimpleNamespace(model_dump=lambda **kw: {})) + runs = control.bind(request) + before = await runs.configuration_revision(assistant_ids=["lead_agent"], durable=True) + data["exempt_tools"] = OrderedSet(["present_files", "task"]) + assert before == await runs.configuration_revision(assistant_ids=["lead_agent"], durable=True) + data["tools"].reverse() + assert before != await runs.configuration_revision(assistant_ids=["lead_agent"], durable=True) diff --git a/backend/tests/test_extension_api_contracts.py b/backend/tests/test_extension_api_contracts.py index 418d88352b..d3152e2b87 100644 --- a/backend/tests/test_extension_api_contracts.py +++ b/backend/tests/test_extension_api_contracts.py @@ -342,7 +342,7 @@ def test_runtime_api_version_matches_the_installed_contract_package(): """Every additive contract slice bumps both gates together.""" from importlib.metadata import version - assert API_VERSION == "0.2.8" + assert API_VERSION == "0.2.9" assert API_VERSION == version("deerflow-extension-api") diff --git a/backend/tests/test_extension_execution_fence.py b/backend/tests/test_extension_execution_fence.py new file mode 100644 index 0000000000..28c497db71 --- /dev/null +++ b/backend/tests/test_extension_execution_fence.py @@ -0,0 +1,158 @@ +"""Execution ownership must be checked in the child admission transaction.""" + +import pytest +import pytest_asyncio +from deerflow_extension_api import ExecutionFence +from sqlalchemy import insert +from test_run_repository import _cleanup, _make_repo + +from deerflow.persistence.extension_leases.model import ExtensionExecutionLeaseRow + + +@pytest_asyncio.fixture +async def repo(tmp_path): + value = await _make_repo(tmp_path) + try: + yield value + finally: + await _cleanup() + + +@pytest.mark.asyncio +async def test_admission_rejects_stale_extension_epoch(repo): + async with repo._sf() as session: + await session.execute(insert(ExtensionExecutionLeaseRow).values(namespace="test", resource_id="workflow", user_id="alice", holder_id="new", epoch=2, expires_at=9999999999)) + await session.commit() + from deerflow.runtime.runs.manager import ConflictError + + with pytest.raises(ConflictError, match="execution fence"): + await repo.create_thread_operation_atomic("child", thread_id="thread", owner_worker_id="worker", lease_expires_at=None, user_id="alice", execution_fence=ExecutionFence("test", "workflow", "old", 1)) + assert await repo.get("child", user_id="alice") is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("changed", ["holder", "epoch", "user", "expired", "missing"]) +async def test_fence_rejects_wrong_authority(repo, changed): + from deerflow.runtime.runs.manager import ConflictError + + values = dict(namespace="test", resource_id="workflow", user_id="alice", holder_id="owner", epoch=1, expires_at=9999999999) + if changed != "missing": + values.update({"holder_id": "other"} if changed == "holder" else {"epoch": 2} if changed == "epoch" else {"user_id": "bob"} if changed == "user" else {"expires_at": 0}) + async with repo._sf() as session: + await session.execute(insert(ExtensionExecutionLeaseRow).values(**values)) + await session.commit() + with pytest.raises((ValueError, ConflictError), match="execution fence"): + await repo.create_thread_operation_atomic("child", thread_id="thread", owner_worker_id="worker", lease_expires_at=None, user_id="alice", execution_fence=ExecutionFence("test", "workflow", "owner", 1)) + + +@pytest.mark.asyncio +async def test_fence_allows_live_owner(repo): + async with repo._sf() as session: + await session.execute(insert(ExtensionExecutionLeaseRow).values(namespace="test", resource_id="workflow", user_id="alice", holder_id="owner", epoch=1, expires_at=9999999999)) + await session.commit() + row, _ = await repo.create_thread_operation_atomic("child", thread_id="thread", owner_worker_id="worker", lease_expires_at=None, user_id="alice", execution_fence=ExecutionFence("test", "workflow", "owner", 1)) + assert row["run_id"] == "child" + + +@pytest.mark.asyncio +async def test_postgres_takeover_fences_admission_after_lock_wait(): + import asyncio + import os + import uuid + + import psycopg + from psycopg import sql + from sqlalchemy import update + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from deerflow.persistence.run import RunRepository + from deerflow.persistence.run.model import RunChangeClockRow, RunRow + from deerflow.runtime.runs.manager import ConflictError + + dsn = os.getenv("DEERFLOW_TEST_WORKFLOW_POSTGRES") + if not dsn: + pytest.skip("Requires dedicated PostgreSQL") + schema = "fence_test_" + uuid.uuid4().hex + with psycopg.connect(dsn, autocommit=True) as db: + db.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema))) + engine = create_async_engine(dsn.split("?")[0].replace("postgresql://", "postgresql+asyncpg://"), connect_args={"server_settings": {"search_path": schema}}) + sessions = async_sessionmaker(engine, expire_on_commit=False) + try: + async with engine.begin() as connection: + for table in (ExtensionExecutionLeaseRow.__table__, RunChangeClockRow.__table__, RunRow.__table__): + await connection.run_sync(table.create) + async with sessions() as session: + await session.execute(insert(ExtensionExecutionLeaseRow).values(namespace="test", resource_id="workflow", user_id="alice", holder_id="old", epoch=1, expires_at=9999999999)) + await session.commit() + repository = RunRepository(sessions) + async with sessions() as takeover: + await takeover.execute(update(ExtensionExecutionLeaseRow).values(holder_id="new", epoch=2)) + admission = asyncio.create_task( + repository.create_thread_operation_atomic("child", thread_id="thread", owner_worker_id="worker", lease_expires_at=None, user_id="alice", execution_fence=ExecutionFence("test", "workflow", "old", 1)) + ) + try: + # Observe an actual database lock waiter before releasing the + # takeover; a sleep-only test could miss the vulnerable window. + from sqlalchemy import text + + for _ in range(200): + async with engine.connect() as observer: + waiting = await observer.scalar(text("SELECT count(*) FROM pg_stat_activity WHERE wait_event_type='Lock' AND query LIKE '%extension_execution_leases%'")) + if waiting: + break + await asyncio.sleep(0.01) + assert waiting + await takeover.commit() + with pytest.raises(ConflictError, match="execution fence"): + await admission + finally: + await takeover.rollback() + await asyncio.gather(admission, return_exceptions=True) + assert await repository.get("child", user_id="alice") is None + finally: + await engine.dispose() + with psycopg.connect(dsn, autocommit=True) as db: + db.execute(sql.SQL("DROP SCHEMA {} CASCADE").format(sql.Identifier(schema))) + + +@pytest.mark.asyncio +async def test_fenced_cancel_cannot_cancel_successor_child(repo): + from deerflow.runtime.runs.manager import ConflictError + + async with repo._sf() as session: + await session.execute(insert(ExtensionExecutionLeaseRow).values(namespace="test", resource_id="workflow", user_id="alice", holder_id="new", epoch=2, expires_at=9999999999)) + await session.commit() + await repo.create_thread_operation_atomic("child", thread_id="thread", owner_worker_id="worker", lease_expires_at=None, user_id="alice") + with pytest.raises(ConflictError, match="execution fence"): + await repo.request_cancel("child", action="interrupt", execution_fence=ExecutionFence("test", "workflow", "old", 1), user_id="alice") + assert (await repo.get("child", user_id="alice"))["cancel_action"] is None + assert await repo.request_cancel("child", action="interrupt", execution_fence=ExecutionFence("test", "workflow", "new", 2), user_id="alice") == "interrupt" + + +@pytest.mark.asyncio +async def test_execution_fence_migration_upgrade_and_downgrade(tmp_path): + import asyncio + + import sqlalchemy as sa + from alembic import command + from sqlalchemy.ext.asyncio import create_async_engine + + from deerflow.persistence import bootstrap + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'fence-migration.db'}") + config = bootstrap._get_alembic_config(engine) + try: + await asyncio.to_thread(bootstrap._upgrade, config, "0047_mcp_oauth_tokens") + async with engine.begin() as connection: + await connection.execute(sa.text("DROP TABLE IF EXISTS extension_execution_leases")) + await asyncio.to_thread(bootstrap._upgrade, config, "0048_extension_execution_leases") + async with engine.connect() as connection: + assert "extension_execution_leases" in await connection.run_sync(lambda c: sa.inspect(c).get_table_names()) + await asyncio.to_thread(command.downgrade, config, "0047_mcp_oauth_tokens") + async with engine.connect() as connection: + names = await connection.run_sync(lambda c: sa.inspect(c).get_table_names()) + assert "extension_execution_leases" not in names + assert "mcp_oauth_tokens" in names + assert (await connection.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == "0047_mcp_oauth_tokens" + finally: + await engine.dispose() diff --git a/backend/tests/test_migration_0047_mcp_oauth_tokens.py b/backend/tests/test_migration_0047_mcp_oauth_tokens.py index be565b596d..11a4403923 100644 --- a/backend/tests/test_migration_0047_mcp_oauth_tokens.py +++ b/backend/tests/test_migration_0047_mcp_oauth_tokens.py @@ -1,4 +1,4 @@ -"""Migration 0047: the shared, encrypted ``mcp_oauth_tokens`` table, and the current migration-chain head.""" +"""Migration 0047: the shared, encrypted ``mcp_oauth_tokens`` table, following the Skill host schema.""" from __future__ import annotations @@ -43,10 +43,11 @@ NULLABLE = {"access_token_enc", "token_request_fingerprint", "token_type", "acce pytestmark = pytest.mark.asyncio -async def test_0047_is_the_chain_head(): - assert bootstrap._get_head_revision() == REVISION +async def test_0047_remains_in_the_single_migration_chain(): + assert bootstrap._get_head_revision() == "0048_extension_execution_leases" script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR)) - assert script.get_heads() == [REVISION] + assert script.get_heads() == ["0048_extension_execution_leases"] + assert script.get_revision("0048_extension_execution_leases").down_revision == REVISION assert script.get_revision(REVISION).down_revision == PREVIOUS assert script.get_revision(PREVIOUS).down_revision == "0045_completed_run_evidence" # alembic_version.version_num is VARCHAR(32). @@ -181,6 +182,6 @@ async def test_bootstrap_provisions_the_table_on_an_empty_database(tmp_path): await bootstrap.bootstrap_schema(engine, backend="sqlite") assert set((await _shape(engine))["columns"]) == COLUMNS async with engine.connect() as conn: - assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == REVISION + assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == bootstrap._get_head_revision() finally: await engine.dispose() diff --git a/backend/tests/test_workflow_acceptance_routing.py b/backend/tests/test_workflow_acceptance_routing.py new file mode 100644 index 0000000000..c48eb219ec --- /dev/null +++ b/backend/tests/test_workflow_acceptance_routing.py @@ -0,0 +1,19 @@ +"""nginx retry histories must identify the final Gateway, not each attempt.""" + +import httpx +import pytest +from test_workflow_gateway_acceptance import final_gateway_port + + +@pytest.mark.parametrize( + "history, expected", + [ + ("192.0.2.1:8001", 8001), + ("[2001:db8::1]:8001, 192.0.2.1:8001", 8001), + ("192.0.2.1:8001, [2001:db8::1]:8002, 192.0.2.1:8002", 8002), + ("[2001:db8::1]:8002", 8002), + ], +) +def test_final_gateway_port(history, expected): + response = httpx.Response(200, headers={"X-Acceptance-Upstream": history}) + assert final_gateway_port(response) == expected diff --git a/backend/tests/test_workflow_cluster.py b/backend/tests/test_workflow_cluster.py new file mode 100644 index 0000000000..320b8485bb --- /dev/null +++ b/backend/tests/test_workflow_cluster.py @@ -0,0 +1,505 @@ +"""Real PostgreSQL coordination; opt in with DEERFLOW_TEST_WORKFLOW_POSTGRES.""" + +import asyncio +import os +import uuid +from dataclasses import replace + +import pytest +import pytest_asyncio +import test_workflow_plugin as fixtures +from test_workflow_plugin import context + +plugin = fixtures.plugin +pytestmark = [pytest.mark.asyncio, pytest.mark.skipif(not os.getenv("DEERFLOW_TEST_WORKFLOW_POSTGRES"), reason="Requires dedicated PostgreSQL")] + + +@pytest_asyncio.fixture +async def cluster(plugin): + import psycopg + from deerflow_extension_workflows.service import Workflows + from psycopg import sql + from psycopg.conninfo import make_conninfo + + dsn = os.environ["DEERFLOW_TEST_WORKFLOW_POSTGRES"] + schema = "workflow_test_" + uuid.uuid4().hex + with psycopg.connect(dsn, autocommit=True) as db: + db.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema))) + scoped = make_conninfo(dsn, options=f"-csearch_path={schema}") + with psycopg.connect(scoped) as db: + db.execute( + "CREATE TABLE extension_execution_leases(namespace varchar(96),resource_id varchar(128),user_id varchar(64) NOT NULL,holder_id varchar(128) NOT NULL,epoch bigint NOT NULL,expires_at double precision NOT NULL," + "PRIMARY KEY(namespace,resource_id))" + ) + original_start = plugin[3].start + plugin[3].fences = [] + + async def fenced_start(*, execution_fence, **kwargs): + assert execution_fence.namespace == "community.workflows" + plugin[3].fences.append(execution_fence) + return await original_start(**kwargs) + + original_cancel = plugin[3].cancel + + async def fenced_cancel(*, execution_fence, **kwargs): + assert execution_fence.namespace == "community.workflows" + return await original_cancel(**kwargs) + + plugin[3].cancel = fenced_cancel + plugin[3].start = fenced_start + config = {**plugin[2].config, "storage_url": scoped, "lease_seconds": 10} + a, b = Workflows(config), Workflows(config) + await a.start(None) + await b.start(None) + try: + yield a, b, plugin[3] + finally: + await asyncio.gather(a.stop(), b.stop()) + with psycopg.connect(dsn, autocommit=True) as db: + db.execute(sql.SQL("DROP SCHEMA {} CASCADE").format(sql.Identifier(schema))) + + +async def test_launch_on_a_resume_on_b(cluster): + a, b, runs = cluster + payload = {"request_id": "same", "script": "await pause({stepId:'p'}); return 42;", "args": {}} + first, retry = await asyncio.gather(a.launch(payload, context(runs)), b.launch(payload, context(runs))) + assert first["run_id"] == retry["run_id"] + run_id = first["run_id"] + for _ in range(200): + record = await b.get({"run_id": run_id}, context(runs)) + if record["status"] == "paused": + break + await asyncio.sleep(0.02) + assert record["status"] == "paused" + await b.resume(dict(run_id=run_id, fingerprint=record["fingerprint"], step_id="p", attempt=1), context(runs)) + await asyncio.wait_for(asyncio.gather(a.wait_idle(run_id), b.wait_idle(run_id)), 10) + assert (await b.get({"run_id": run_id}, context(runs)))["result"] == 42 + + +async def test_expired_owner_cannot_write_after_takeover(cluster): + a, b, runs = cluster + payload = {"request_id": "takeover", "script": "await pause({stepId:'p'}); return 42;", "args": {}} + run_id = (await a.launch(payload, context(runs)))["run_id"] + for _ in range(200): + old = await asyncio.to_thread(a.store.get, "alice", run_id) + if old["status"] == "paused": + break + await asyncio.sleep(0.02) + a.heartbeats[run_id].cancel() + await asyncio.gather(a.heartbeats[run_id], return_exceptions=True) + + def expire(): + from contextlib import closing + + with closing(a.store.connect()) as db, db: + db.execute("UPDATE extension_execution_leases SET expires_at=0 WHERE resource_id=?", (run_id,)) + + await asyncio.to_thread(expire) + session = replace(context(runs), interaction="session") + await b.recover({"run_id": run_id}, session) + with pytest.raises(ValueError, match="lease lost"): + await asyncio.to_thread(a.store.put, "alice", old) + # Restore the stale owner after takeover; its next renewal must kill only + # its own worker and leave the successor and shared record untouched. + await a.follow_lease(a.entries[run_id]) + for _ in range(200): + record = await b.get({"run_id": run_id}, session) + if record["status"] == "paused": + break + await asyncio.sleep(0.02) + await b.resume(dict(run_id=run_id, fingerprint=record["fingerprint"], step_id="p", attempt=1), session) + await asyncio.wait_for(b.wait_idle(run_id), 10) + assert (await b.get({"run_id": run_id}, session))["result"] == 42 + + +async def test_shared_workflow_limit(cluster): + a, b, runs = cluster + a.config["max_workflows"] = b.config["max_workflows"] = 1 + await a.launch({"request_id": "first", "script": "await pause({stepId:'p'});", "args": {}}, context(runs)) + with pytest.raises(ValueError, match="workflow limit"): + await b.launch({"request_id": "second", "script": "return 1;", "args": {}}, context(runs, owner="bob")) + + +async def poll(service, runs, run_id, status): + for _ in range(300): + record = await service.get({"run_id": run_id}, context(runs)) + if record["status"] == status: + return record + await asyncio.sleep(0.02) + raise AssertionError(record) + + +async def test_cross_gateway_approval_is_exact_and_idempotent(cluster): + a, b, runs = cluster + run_id = ( + await a.launch( + { + "request_id": "approval", + "script": "const a=await agent({stepId:'a',agentRef:'lead_agent',task:1}); await humanCheckpoint({stepId:'c',sourceStepId:'a',artifactVersion:a.artifactVersion}); " + "return await agent({stepId:'b',agentRef:'reviewer',task:2});", + "args": {}, + }, + context(runs), + ) + )["run_id"] + record = await poll(b, runs, run_id, "waiting_approval") + session = replace(context(runs), interaction="session") + payload = dict(run_id=run_id, step_id="c", attempt=1, artifact_version=record["checkpoints"][0]["artifact_version"], decision="approve") + with pytest.raises(ValueError, match="exact checkpoint"): + await b.decide({**payload, "artifact_version": "wrong"}, session) + await asyncio.gather(a.decide(payload, session), b.decide(payload, session)) + await asyncio.wait_for(a.wait_idle(run_id), 10) + assert (await b.get({"run_id": run_id}, session))["status"] == "completed" + assert len(runs.starts) == 2 + assert len(runs.fences) == 2 + + +async def test_cross_gateway_cancel_and_revocation(cluster): + a, b, runs = cluster + run_id = (await a.launch({"request_id": "cancel", "script": "await pause({stepId:'p'}); return 1;", "args": {}}, context(runs)))["run_id"] + record = await poll(b, runs, run_id, "paused") + runs.denied = True + with pytest.raises(Exception, match="revoked"): + await b.resume(dict(run_id=run_id, fingerprint=record["fingerprint"], step_id="p", attempt=1), context(runs)) + runs.denied = False + await b.cancel({"run_id": run_id}, context(runs)) + await asyncio.wait_for(a.wait_idle(run_id), 10) + assert (await b.get({"run_id": run_id}, context(runs)))["status"] == "cancelled" + + +async def test_real_owner_kill_during_review_and_rolling_upgrade(cluster): + import json + import sys + from pathlib import Path + + a, b, runs = cluster + root = Path(__file__).resolve().parents[2] + program = r""" +import asyncio,json,sys +from deerflow_extension_workflows.service import Workflows +from test_workflow_plugin import Runs,context +async def main(): + service=Workflows(json.loads(sys.argv[1]));await service.start(None) + runs=Runs() + result=await service.launch({'request_id':'killed-owner','script':"await pause({stepId:'p'}); return 7;",'args':{}},context(runs)) + while (await service.get({'run_id':result['run_id']},context(runs)))['status']!='paused': + await asyncio.sleep(.01) + print(result['run_id'],flush=True) + await asyncio.Event().wait() +asyncio.run(main()) +""" + child = await asyncio.create_subprocess_exec( + sys.executable, + "-c", + program, + json.dumps(a.config), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env={**os.environ, "PYTHONPATH": os.pathsep.join([str(root / "examples/deerflow-extension-workflows"), str(root / "backend/tests"), str(root / "backend")])}, + ) + try: + run_id = (await asyncio.wait_for(child.stdout.readline(), 10)).decode().strip() + assert len(run_id) == 32 + finally: + child.kill() + await child.wait() + + def expire(): + from contextlib import closing + + with closing(b.store.connect()) as db, db: + db.execute("UPDATE extension_execution_leases SET expires_at=0 WHERE resource_id=?", (run_id,)) + + await asyncio.to_thread(expire) + session = replace(context(runs), interaction="session") + version = b.runtime_version + b.runtime_version = "incompatible-upgrade" + with pytest.raises(ValueError, match="runtime version"): + await b.recover({"run_id": run_id}, session) + b.runtime_version = version + runs.denied = True + with pytest.raises(Exception, match="revoked"): + await b.recover({"run_id": run_id}, session) + runs.denied = False + await b.recover({"run_id": run_id}, session) + record = await poll(b, runs, run_id, "paused") + await b.resume(dict(run_id=run_id, fingerprint=record["fingerprint"], step_id="p", attempt=1), session) + await asyncio.wait_for(b.wait_idle(run_id), 10) + assert (await b.get({"run_id": run_id}, session))["result"] == 7 + + +async def test_cancel_keeps_lease_until_child_cleanup_finishes(cluster, monkeypatch): + a, b, runs = cluster + cleaning = asyncio.Event() + finish_cleanup = asyncio.Event() + original_cancel = a.cancel_child + + async def held_cancel(*args, **kwargs): + cleaning.set() + await finish_cleanup.wait() + return await original_cancel(*args, **kwargs) + + monkeypatch.setattr(a, "cancel_child", held_cancel) + script = "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});" + run_id = (await a.launch({"request_id": "cleanup-lease", "script": script, "args": {}}, context(runs)))["run_id"] + await asyncio.wait_for(runs.entered.wait(), 5) + try: + # Agent admission signals before the supervisor persists its run ID. + # Wait for that commit so remote cancel does not race its revision CAS. + async with asyncio.timeout(5): + while True: + record = await b.get({"run_id": run_id}, context(runs)) + if record["steps"][0]["status"] == "running": + break + await asyncio.sleep(0.01) + await b.cancel({"run_id": run_id}, context(runs)) + await asyncio.wait_for(cleaning.wait(), 5) + # The supervisor has stopped its worker but is still draining the child. + async with asyncio.timeout(5): + while run_id in a.processes: + await asyncio.sleep(0.01) + finally: + finish_cleanup.set() + await asyncio.wait_for(a.wait_idle(run_id), 10) + record = await b.get({"run_id": run_id}, context(runs)) + assert record["status"] == "cancelled" + assert record["steps"][0]["status"] == "cancelled" + assert not a.entries + + +async def test_shared_child_quota_and_cancel(cluster): + a, b, runs = cluster + a.config["global_concurrency"] = b.config["global_concurrency"] = 1 + script = "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});" + first = (await a.launch({"request_id": "slot-a", "script": script, "args": {}}, context(runs)))["run_id"] + await asyncio.wait_for(runs.entered.wait(), 5) + second = (await b.launch({"request_id": "slot-b", "script": script, "args": {}}, context(runs)))["run_id"] + await asyncio.sleep(0.2) + assert len(runs.starts) == 1 + await b.cancel({"run_id": first}, context(runs)) + await asyncio.wait_for(a.wait_idle(first), 10) + await asyncio.wait_for(b.wait_idle(second), 10) + assert len(runs.starts) == 2 + assert runs.peak == 1 + + +async def test_capture_is_fenced_and_immutable(cluster): + a, b, runs = cluster + run_id = (await a.launch({"request_id": "file", "script": "await pause({stepId:'p'});", "args": {}}, context(runs)))["run_id"] + await poll(a, runs, run_id, "paused") + record = await asyncio.to_thread(a.store.get, "alice", run_id) + with pytest.raises(ValueError, match="lease lost"): + await asyncio.to_thread(b.store.capture_artifact, "alice", run_id, "file", "report.txt", b"wrong", execution_record=record) + info = await asyncio.to_thread(a.store.capture_artifact, "alice", run_id, "file", "report.txt", b"original", execution_record=record) + assert (await asyncio.to_thread(b.store.artifact, "alice", run_id, info["version"]))[1] == b"original" + with pytest.raises(ValueError, match="overwritten"): + await asyncio.to_thread(a.store.capture_artifact, "alice", run_id, "file", "report.txt", b"changed", execution_record=record) + + +@pytest.mark.parametrize("boundary", ["admission", "result", "approval"]) +async def test_cross_gateway_recovery_persistence_boundaries(cluster, monkeypatch, boundary): + import copy + import json + from contextlib import closing + + a, b, runs = cluster + snapshots = [] + original = runs.start if boundary == "admission" else runs.configuration_evidence + + async def capture(**kwargs): + value = await original(**kwargs) + entry = next(iter(a.entries.values())) + snapshots.append(await asyncio.to_thread(a.store.get, "alice", entry["record"]["id"])) + return value + + if boundary != "approval": + monkeypatch.setattr(runs, "start" if boundary == "admission" else "configuration_evidence", capture) + script = "const a=await agent({stepId:'a',agentRef:'lead_agent',task:1}); await humanCheckpoint({stepId:'c',sourceStepId:'a',artifactVersion:a.artifactVersion}); return await agent({stepId:'b',agentRef:'reviewer',task:2});" + run_id = (await a.launch({"request_id": boundary, "script": script, "args": {}}, context(runs)))["run_id"] + record = await poll(a, runs, run_id, "waiting_approval") + session = replace(context(runs), interaction="session") + if boundary == "approval": + a.heartbeats[run_id].cancel() + await asyncio.gather(a.heartbeats[run_id], return_exceptions=True) + await b.decide(dict(run_id=run_id, step_id="c", attempt=1, artifact_version=record["checkpoints"][0]["artifact_version"], decision="approve"), session) + snapshots.append(await asyncio.to_thread(a.store.get, "alice", run_id)) + snapshot = copy.deepcopy(snapshots[0]) + if boundary != "approval": + monkeypatch.setattr(runs, "start" if boundary == "admission" else "configuration_evidence", original) + await a.stop() + + def install_crash_image(): + with closing(b.store.connect()) as db, db: + revision = db.execute("SELECT revision FROM workflows WHERE id=?", (run_id,)).fetchone()[0] + 1 + snapshot["revision"] = revision + db.execute("UPDATE workflows SET data=?,status=?,revision=? WHERE id=?", (json.dumps(snapshot), snapshot["status"], revision, run_id)) + db.execute("UPDATE extension_execution_leases SET expires_at=0 WHERE resource_id=?", (run_id,)) + + await asyncio.to_thread(install_crash_image) + restored = await b.recover({"run_id": run_id}, session) + if boundary == "admission": + assert restored["status"] == "needs_reconciliation" + assert len(runs.starts) == 1 + else: + if boundary == "result": + record = await poll(b, runs, run_id, "waiting_approval") + await b.decide(dict(run_id=run_id, step_id="c", attempt=1, artifact_version=record["checkpoints"][0]["artifact_version"], decision="approve"), session) + await asyncio.wait_for(b.wait_idle(run_id), 10) + assert (await b.get({"run_id": run_id}, session))["status"] == "completed" + assert len(runs.starts) == 2 + + +async def test_expired_cancel_command_cannot_resume_dispatch(cluster): + from contextlib import closing + + a, b, runs = cluster + run_id = (await a.launch({"request_id": "cancel-before-crash", "script": "await pause({stepId:'p'}); return 1;", "args": {}}, context(runs)))["run_id"] + await poll(a, runs, run_id, "paused") + a.heartbeats[run_id].cancel() + await asyncio.gather(a.heartbeats[run_id], return_exceptions=True) + await b.cancel({"run_id": run_id}, context(runs)) + + def expire(): + with closing(b.store.connect()) as db, db: + db.execute("UPDATE extension_execution_leases SET expires_at=0 WHERE resource_id=?", (run_id,)) + + await asyncio.to_thread(expire) + session = replace(context(runs), interaction="session") + with pytest.raises(ValueError, match="Cancellation pending"): + await b.recover({"run_id": run_id}, session) + await a.follow_lease(a.entries[run_id]) + await b.cancel({"run_id": run_id}, session) + assert (await b.get({"run_id": run_id}, session))["status"] == "cancelled" + + +async def test_cluster_policy_rejects_mismatched_replica(cluster): + from deerflow_extension_workflows.service import Workflows + + a, _, _ = cluster + peer = Workflows({**a.config, "global_concurrency": 1}) + with pytest.raises(ValueError, match="cluster limits differ"): + await peer.start(None) + + +async def test_peer_rejects_configuration_change_and_expired_review(cluster): + import json + from contextlib import closing + + a, b, runs = cluster + run_id = (await a.launch({"request_id": "config", "script": "await pause({stepId:'p'}); return 1;", "args": {}}, context(runs)))["run_id"] + record = await poll(a, runs, run_id, "paused") + payload = dict(run_id=run_id, fingerprint=record["fingerprint"], step_id="p", attempt=1) + runs.revision = "changed-config" + with pytest.raises(ValueError, match="configuration changed"): + await b.resume(payload, context(runs)) + runs.revision = "initial" + a.heartbeats[run_id].cancel() + await asyncio.gather(a.heartbeats[run_id], return_exceptions=True) + + def expire_checkpoint(): + with closing(b.store.connect()) as db, db: + retained = json.loads(db.execute("SELECT data FROM workflows WHERE id=?", (run_id,)).fetchone()[0]) + retained["pause_expires_at"] = 0 + db.execute("UPDATE workflows SET data=? WHERE id=?", (json.dumps(retained), run_id)) + + await asyncio.to_thread(expire_checkpoint) + with pytest.raises(ValueError, match="expired"): + await b.resume(payload, context(runs)) + + +@pytest.mark.parametrize("saturated", [False, True]) +async def test_expired_cancelling_can_finish_cleanup_and_delete(cluster, saturated): + import copy + from contextlib import closing + + a, b, runs = cluster + run_id = (await a.launch({"request_id": "cancel-recovery", "script": "await pause({stepId:'p'}); return 1;", "args": {}}, context(runs)))["run_id"] + await poll(a, runs, run_id, "paused") + await a.stop() + + # Crash image of cancel_recovered after persisting cancelling, before cleanup. + def crash(): + with closing(b.store.connect()) as db, db: + import json + + record = json.loads(db.execute("SELECT data FROM workflows WHERE id=?", (run_id,)).fetchone()[0]) + record["status"] = "cancelling" + db.execute("UPDATE workflows SET status='cancelling',data=? WHERE id=?", (json.dumps(record), run_id)) + db.execute("UPDATE extension_execution_leases SET expires_at=0 WHERE resource_id=?", (run_id,)) + if saturated: + # Other active rows already consume both budget ceilings. + for i in range(max(b.config["max_workflows"], b.config["max_workflows_per_owner"])): + other = copy.deepcopy(record) + other.update(id=f"occupied-{i}", status="running") + db.execute("INSERT INTO workflows(id,owner,request_id,fingerprint,status,data) VALUES (?,?,?,?,?,?)", (other["id"], "alice", other["id"], "occupied", "running", json.dumps(other))) + + await asyncio.to_thread(crash) + session = replace(context(runs), interaction="session") + with pytest.raises(ValueError, match="Cancellation pending"): + await b.recover({"run_id": run_id}, session) + result = await b.cancel({"run_id": run_id}, session) + assert result["status"] == "cancelled" + await b.delete({"run_id": run_id}, session) + with pytest.raises(ValueError, match="unavailable"): + await b.get({"run_id": run_id}, session) + assert not runs.starts + + +async def test_reads_do_not_wait_for_cluster_budget_lock(cluster): + import psycopg + from deerflow_extension_workflows.postgres import LOCK + + a, b, runs = cluster + run_id = (await a.launch({"request_id": "read-lock", "script": "return 1;", "args": {}}, context(runs)))["run_id"] + await poll(a, runs, run_id, "completed") + with psycopg.connect(b.store.dsn) as blocker: + blocker.execute("SELECT pg_advisory_xact_lock(%s)", (LOCK,)) + # A real held advisory lock makes the old read path hit lock_timeout. + assert (await asyncio.to_thread(b.store.get, "alice", run_id))["status"] == "completed" + assert [r["id"] for r in await asyncio.to_thread(b.store.list, "alice")] == [run_id] + assert await asyncio.to_thread(b.store.list, "other") == [] + + +@pytest.mark.parametrize("failure", ["permission", "configuration", "uncertain"]) +async def test_rejected_admission_slot_accounting(cluster, monkeypatch, failure): + from contextlib import closing + + from deerflow_extension_api import AgentRunError + + a, b, runs = cluster + a.config["global_concurrency"] = b.config["global_concurrency"] = 1 + original_start = runs.start + original_check = a.check_configuration + + async def reject_start(**kwargs): + raise AgentRunError(403 if failure == "permission" else 503, "synthetic rejection") + + async def reject_step_configuration(entry): + if entry["record"]["steps"]: + raise ValueError("Configuration changed") + await original_check(entry) + + if failure == "configuration": + monkeypatch.setattr(a, "check_configuration", reject_step_configuration) + else: + monkeypatch.setattr(runs, "start", reject_start) + script = "return await agent({stepId:'a',agentRef:'lead_agent',task:'hello'});" + run_id = (await a.launch({"request_id": "rejected", "script": script, "args": {}}, context(runs)))["run_id"] + await asyncio.wait_for(a.wait_idle(run_id), 10) + assert not runs.starts + + def slots(): + with closing(a.store.connect(readonly=True)) as db: + return db.execute("SELECT count(*) FROM workflow_slots").fetchone()[0] + + assert await asyncio.to_thread(slots) == (1 if failure == "uncertain" else 0) + if failure == "uncertain": + with pytest.raises(ValueError, match="Reconcile retained"): + await b.delete({"run_id": run_id}, context(runs)) + return + await b.delete({"run_id": run_id}, context(runs)) + monkeypatch.setattr(runs, "start", original_start) + peer = (await b.launch({"request_id": "peer-after-rejection", "script": script, "args": {}}, context(runs)))["run_id"] + await asyncio.wait_for(b.wait_idle(peer), 10) + assert (await b.get({"run_id": peer}, context(runs)))["status"] == "completed" + assert len(runs.starts) == 1 diff --git a/backend/tests/test_workflow_gateway_acceptance.py b/backend/tests/test_workflow_gateway_acceptance.py new file mode 100644 index 0000000000..80b4011843 --- /dev/null +++ b/backend/tests/test_workflow_gateway_acceptance.py @@ -0,0 +1,581 @@ +"""Opt-in real two-Gateway acceptance with disposable Docker Postgres/Redis/nginx. + +Run with DEERFLOW_WORKFLOW_ACCEPTANCE=1 and a working Docker CLI. Only model +answers and fault delays are synthetic. All HTTP/auth/host/store paths are real. +""" + +import base64 +import json +import os +import shutil +import signal +import socket +import subprocess +import sys +import time +import uuid +from pathlib import Path + +import httpx +import pytest +import yaml + +ROOT = Path(__file__).resolve().parents[2] +EXAMPLE = ROOT / "examples/deerflow-extension-workflows" +pytestmark = pytest.mark.skipif(os.name != "posix" or os.getenv("DEERFLOW_WORKFLOW_ACCEPTANCE") != "1", reason="Opt-in POSIX two-Gateway Docker acceptance") + + +def until(fn, seconds=60): + end = time.monotonic() + seconds + while time.monotonic() < end: + value = fn() + if value: + return value + time.sleep(0.1) + raise AssertionError(f"Timed out waiting for {fn}") + + +def port(): + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class Cluster: + def __init__(self, root): + self.root = root + self.docker = os.getenv("DEERFLOW_TEST_DOCKER", "docker") + self.containers = [] + self.processes = {} + self.logs = [] + self.ports = {key: port() for key in ("a", "b", "proxy", "pg", "redis")} + self.http = httpx.Client(timeout=35, trust_env=False) + self.cookies = {} + + def docker_run(self, image, *args, options=()): + name = "workflow-acceptance-" + uuid.uuid4().hex[:12] + self.containers.append(name) + subprocess.run([self.docker, "run", "--detach", "--name", name, *options, image, *args], check=True, capture_output=True, text=True) + return name + + def setup(self): + import importlib.util + + spec = importlib.util.spec_from_file_location("workflow_multi_helper", ROOT / "scripts/dev_multi_instance.py") + helper = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = helper + try: + spec.loader.exec_module(helper) + finally: + sys.modules.pop(spec.name, None) + build_multi_instance_config, generate_shared_secrets = helper.build_multi_instance_config, helper.generate_shared_secrets + + pg = self.docker_run("postgres:17-alpine", options=("-e", "POSTGRES_HOST_AUTH_METHOD=trust", "-p", f"127.0.0.1:{self.ports['pg']}:5432")) + self.docker_run("redis:7-alpine", options=("-p", f"127.0.0.1:{self.ports['redis']}:6379")) + until(lambda: subprocess.run([self.docker, "exec", pg, "pg_isready", "-U", "postgres"], capture_output=True).returncode == 0) + self.dsn = f"postgresql://postgres@127.0.0.1:{self.ports['pg']}/postgres" + config, _ = build_multi_instance_config({}, database_url=self.dsn, redis_url=f"redis://127.0.0.1:{self.ports['redis']}/0", sandbox_container_prefix="workflow-acceptance") + config.update( + config_version=61, + models=[{"name": "acceptance", "use": "support.workflow_gateway:AcceptanceModel", "model": "synthetic", "supports_tool_calling": True}], + sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + tools=[{"name": "write_file", "group": "file:write", "use": "deerflow.sandbox.tools:write_file_tool"}], + tool_groups=[{"name": "file:write"}], + title={"enabled": False}, + memory={"enabled": False}, + summarization={"enabled": False}, + plugins=[ + { + "name": "workflows", + "use": "deerflow_extension_workflows:install", + "enabled": True, + "config": { + "enabled": True, + "agents": ["lead_agent"], + "storage_path": str(self.root / "workflows.sqlite"), + "storage_url": self.dsn, + "lease_seconds": 5, + "max_workflows": 8, + "max_workflows_per_owner": 4, + "timeout_seconds": 120, + }, + } + ], + ) + self.config = config + (self.root / "config.yaml").write_text(yaml.safe_dump(config)) + (self.root / "extensions.json").write_text("{}") + self.env = { + **os.environ, + **generate_shared_secrets(), + "DEER_FLOW_CONFIG_PATH": str(self.root / "config.yaml"), + "DEER_FLOW_EXTENSIONS_CONFIG_PATH": str(self.root / "extensions.json"), + "DEER_FLOW_HOME": str(self.root / "home"), + "DEER_FLOW_PROJECT_ROOT": str(ROOT), + "DEER_FLOW_ENV_FILE": str(self.root / "empty.env"), + "DEER_FLOW_AUTH_DISABLED": "0", + "GATEWAY_WORKERS": "1", + "WEB_CONCURRENCY": "1", + } + (self.root / "empty.env").touch() + self.start("a") + self.start("b") + # Linux host networking keeps both Gateway listeners loopback-only. + # OrbStack/Docker Desktop exposes host loopback via host.docker.internal. + linux = sys.platform == "linux" + upstream = "127.0.0.1" if linux else "host.docker.internal" + listen = f"127.0.0.1:{self.ports['proxy']}" if linux else "80" + conf = ( + "events {}\nhttp { upstream gateways { " + + " ".join(f"server {upstream}:{self.ports[k]};" for k in ("a", "b")) + + f" }} server {{ listen {listen};" + + " location / { proxy_pass http://gateways; proxy_buffering off; add_header X-Acceptance-Upstream $upstream_addr always; } } }" + ) + (self.root / "nginx.conf").write_text(conf) + network = ("--network", "host") if linux else ("-p", f"127.0.0.1:{self.ports['proxy']}:80", "--add-host", "host.docker.internal:host-gateway") + self.docker_run("nginx:alpine", options=(*network, "-v", f"{self.root / 'nginx.conf'}:/etc/nginx/nginx.conf:ro")) + until(lambda: self.healthy("proxy")) + self.credentials = {"email": "acceptance@example.com", "password": uuid.uuid4().hex} + response = self.request("a", "POST", "/api/v1/auth/initialize", json=self.credentials, expected=201) + self.cookies = dict(response.cookies) + + def start(self, name, extension=EXAMPLE): + fault = self.root / name + fault.mkdir(exist_ok=True) + for item in fault.iterdir(): + item.unlink() + env = {**self.env, "WORKFLOW_FAULT_DIR": str(fault), "PYTHONHASHSEED": "1" if name == "a" else "2", "PYTHONPATH": os.pathsep.join((str(extension), str(ROOT / "backend/tests"), str(ROOT / "backend")))} + log = (self.root / f"{name}.log").open("a") + self.logs.append(log) + self.processes[name] = subprocess.Popen([sys.executable, str(ROOT / "backend/tests/support/workflow_gateway.py"), str(self.ports[name])], cwd=ROOT / "backend", env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + + def ready(): + if self.processes[name].poll() is not None: + raise AssertionError((self.root / f"{name}.log").read_text()[-10000:]) + return self.healthy(name) + + until(ready, 120) + + def healthy(self, name): + try: + return self.http.get(f"http://127.0.0.1:{self.ports[name]}/health").status_code == 200 + except httpx.TransportError: + return False + + def kill(self, name): + p = self.processes.pop(name, None) + if p and p.poll() is None: + os.killpg(p.pid, signal.SIGKILL) + p.wait(timeout=10) + + def close(self): + for name in list(self.processes): + self.kill(name) + self.http.close() + for log in self.logs: + log.close() + for name in self.containers: + subprocess.run([self.docker, "rm", "-f", "-v", name], capture_output=True) + + def request(self, target, method, path, expected=200, **kwargs): + headers = {"Cookie": "; ".join(f"{k}={v}" for k, v in self.cookies.items())} + if self.cookies.get("csrf_token"): + headers["X-CSRF-Token"] = self.cookies["csrf_token"] + r = self.http.request(method, f"http://127.0.0.1:{self.ports[target]}{path}", headers=headers, **kwargs) + self.http.cookies.clear() + assert r.status_code == expected, (r.status_code, r.text) + return r + + def action(self, target, name, payload, expected=200): + return self.request(target, "POST", f"/api/plugins/community.workflows/actions/{name}", json=payload, expected=expected).json() + + def launch(self, script, target="a"): + return self.action(target, "launch", {"request_id": uuid.uuid4().hex, "script": script, "args": {}})["run_id"] + + def get(self, run_id, target="b"): + return self.action(target, "get", {"run_id": run_id}) + + def wait(self, run_id, status, target="b"): + def check(): + record = self.get(run_id, target) + if record["status"] in ("failed", "needs_reconciliation") and status != record["status"]: + raise AssertionError((record, self.sql("SELECT run_id,status,error FROM runs WHERE thread_id=ANY(%s)", ([s["thread_id"] for s in record["steps"]],)))) + return record if record["status"] == status else None + + return until(check) + + def arm(self, boundary, target="a"): + for name in ("reached.json", "release"): + (self.root / target / name).unlink(missing_ok=True) + (self.root / target / "arm.json").write_text(json.dumps({"boundary": boundary})) + + def reached(self, target="a"): + until(lambda: (self.root / target / "reached.json").exists()) + + def recover(self, run_id): + self.wait(run_id, "recoverable") + return self.action("b", "recover", {"run_id": run_id}) + + def resume(self, run_id, target="b"): + record = self.wait(run_id, "paused", target) + return self.action(target, "resume", {"run_id": run_id, "fingerprint": record["fingerprint"], "step_id": record["checkpoints"][-1]["step_id"], "attempt": 1}) + + def decide(self, run_id, target="b"): + record = self.wait(run_id, "waiting_approval", target) + checkpoint = record["checkpoints"][-1] + payload = {"run_id": run_id, "step_id": checkpoint["step_id"], "attempt": 1, "artifact_version": checkpoint["artifact_version"], "decision": "approve"} + self.action(target, "decide", payload) + return payload + + def sql(self, statement, params=()): + import psycopg + + with psycopg.connect(self.dsn) as db: + return db.execute(statement, params).fetchall() + + +@pytest.fixture(scope="module") +def cluster(tmp_path_factory): + c = Cluster(tmp_path_factory.mktemp("workflow-gateway")) + print(f"Acceptance logs: {c.root}") + try: + c.setup() + yield c + finally: + c.close() + + +def final_gateway_port(response): + # nginx records all attempts, including failed IPv6 -> IPv4 retries. + destination = response.headers["x-acceptance-upstream"].rsplit(",", 1)[-1].strip() + return int(destination.rsplit(":", 1)[-1]) + + +def test_authenticated_cross_gateway_control_without_affinity(cluster): + c = cluster + for target in ("a", "b"): + c.request(target, "GET", "/health/ready") + upstreams = {final_gateway_port(c.request("proxy", "GET", "/api/plugins")) for _ in range(6)} + assert upstreams == {c.ports["a"], c.ports["b"]} + run = c.launch("await pause({stepId:'p'}); return await agent({stepId:'child',agentRef:'lead_agent',task:'hello'});") + routed = [c.request("proxy", "POST", "/api/plugins/community.workflows/actions/get", json={"run_id": run}) for _ in range(6)] + assert {r.json()["run_id"] for r in routed} == {run} + assert {final_gateway_port(r) for r in routed} == {c.ports["a"], c.ports["b"]} + c.resume(run) + result = c.wait(run, "completed") + assert len(result["steps"]) == 1 + assert result["steps"][0]["configuration_version"] + run = c.launch("await pause({stepId:'p'}); return 1;") + c.wait(run, "paused") + c.action("b", "cancel", {"run_id": run}) + c.wait(run, "cancelled") + + +@pytest.mark.parametrize("boundary", ["admission", "result_before", "result", "approval"]) +def test_sigkill_at_persistence_boundaries(cluster, boundary): + c = cluster + c.arm(boundary) + script = "const a=await agent({stepId:'first',agentRef:'lead_agent',task:'first'});" + if boundary == "approval": + script += "await humanCheckpoint({stepId:'review',sourceStepId:'first',artifactVersion:a.artifactVersion});" + script += "return await agent({stepId:'second',agentRef:'lead_agent',task:'second'});" + run = c.launch(script) + if boundary == "approval": + c.decide(run) + c.reached() + # Confirm the real boundary before killing, then inspect the shared host DB. + before = c.get(run) + thread = before["steps"][0]["thread_id"] + assert c.sql("SELECT count(*) FROM runs WHERE thread_id=%s", (thread,))[0][0] == 1 + if boundary == "admission": + assert before["steps"][0]["run_id"] is None + if boundary == "result": + assert before["steps"][0]["status"] == "completed" + if boundary == "approval": + assert before["checkpoints"][0]["status"] == "approved" + c.kill("a") + try: + recovered = c.recover(run) + if boundary == "admission": + assert recovered["status"] == "needs_reconciliation" + assert len(c.get(run)["steps"]) == 1 + assert c.sql("SELECT count(*) FROM workflow_slots WHERE workflow=%s", (run,))[0][0] == 1 + else: + record = c.wait(run, "completed") + assert len(record["steps"]) == 2 + if boundary == "approval": + assert record["checkpoints"][0]["status"] == "approved" + assert c.sql("SELECT count(*) FROM runs WHERE thread_id=%s", (thread,))[0][0] == 1 + finally: + c.start("a") + + +def test_file_approval_survives_owner_death_and_source_mutation(cluster): + from support.workflow_gateway import EVIDENCE + + c = cluster + run = c.launch( + "await agent({stepId:'writer',agentRef:'lead_agent',task:'CREATE_EVIDENCE'});" + "const f=await captureArtifact({stepId:'file',sourceStepId:'writer',path:'/mnt/user-data/outputs/evidence.txt'});" + "await humanCheckpoint({stepId:'review',sourceStepId:'file',artifactVersion:f.artifactVersion});" + "const bytes=await readArtifact({artifactVersion:f.artifactVersion});" + "return await agent({stepId:'reader',agentRef:'lead_agent',task:bytes});" + ) + record = c.wait(run, "waiting_approval") + version = record["checkpoints"][0]["artifact_version"] + chunk = c.action("b", "get_artifact", {"run_id": run, "version": version}) + assert base64.b64decode(chunk["content_base64"]) == EVIDENCE.encode() + files = list((c.root / "home").glob("**/outputs/evidence.txt")) + assert len(files) == 1 + files[0].write_text("MUTATED after capture") + c.kill("a") + try: + c.recover(run) + c.decide(run) + record = c.wait(run, "completed") + reader = record["result"] + # The real model echoes the downstream human input; verify exact captured bytes. + payload = json.loads(reader["output"]["echo"]) + assert base64.b64decode(payload["content_base64"]) == EVIDENCE.encode() + assert payload["version"] == version + assert len(record["steps"]) == 2 + finally: + c.start("a") + + +def test_sigstop_stale_owner_cannot_admit_after_takeover(cluster): + c = cluster + c.arm("before_admission") + run = c.launch("return await agent({stepId:'first',agentRef:'lead_agent',task:'stale'});") + c.reached() + record = c.get(run) + thread = record["steps"][0]["thread_id"] + assert c.sql("SELECT count(*) FROM runs WHERE thread_id=%s", (thread,))[0][0] == 0 + p = c.processes["a"] + os.kill(p.pid, signal.SIGSTOP) + try: + assert c.recover(run)["status"] == "needs_reconciliation" + after = c.sql("SELECT data FROM workflows WHERE id=%s", (run,))[0][0] + (c.root / "a/release").touch() + finally: + os.kill(p.pid, signal.SIGCONT) + until(lambda: (c.root / "a/admission_error").exists()) + assert c.sql("SELECT count(*) FROM runs WHERE thread_id=%s", (thread,))[0][0] == 0 + assert c.sql("SELECT data FROM workflows WHERE id=%s", (run,))[0][0] == after + c.kill("a") + c.start("a") + + +def test_distinct_runtime_rollout_drains_old_and_rejects_incompatible_takeover(cluster): + c = cluster + alternate = c.root / "alternate" + shutil.copytree(EXAMPLE, alternate, ignore=shutil.ignore_patterns("__pycache__")) + runtime = alternate / "deerflow_extension_workflows/worker.py" + runtime.write_text(runtime.read_text().replace("MAX_FRAME = 48 * 1024", "MAX_FRAME = 47 * 1024")) + c.kill("b") + c.start("b", alternate) + new_run = c.launch("return 11;", "b") + assert c.wait(new_run, "completed")["result"] == 11 + run = c.launch("await pause({stepId:'p'}); return 17;") + c.wait(run, "paused") + assert c.sql("SELECT data::jsonb->>'runtime_version' FROM workflows WHERE id=%s", (run,)) != c.sql("SELECT data::jsonb->>'runtime_version' FROM workflows WHERE id=%s", (new_run,)) + c.resume(run) # New replica can control the old live runtime; owner A drains it. + assert c.wait(run, "completed")["result"] == 17 + run = c.launch("await pause({stepId:'p'}); return 19;") + c.wait(run, "paused") + c.kill("a") + try: + c.wait(run, "recoverable") + before = c.sql("SELECT epoch FROM extension_execution_leases WHERE resource_id=%s", (run,)) + c.action("b", "recover", {"run_id": run}, expected=422) + assert c.sql("SELECT epoch FROM extension_execution_leases WHERE resource_id=%s", (run,)) == before + finally: + c.start("a") + c.kill("b") + c.start("b") + c.recover(run) + c.resume(run) + assert c.wait(run, "completed")["result"] == 19 + + +def test_concurrent_duplicate_launch_and_approval(cluster): + from concurrent.futures import ThreadPoolExecutor + + c = cluster + payload = { + "request_id": uuid.uuid4().hex, + "args": {}, + "script": ( + "const a=await agent({stepId:'a',agentRef:'lead_agent',task:'first'});" + "await humanCheckpoint({stepId:'review',sourceStepId:'a',artifactVersion:a.artifactVersion});" + "return await agent({stepId:'b',agentRef:'lead_agent',task:'second'});" + ), + } + with ThreadPoolExecutor(2) as pool: + ids = list(pool.map(lambda target: c.action(target, "launch", payload)["run_id"], ("a", "b"))) + assert ids[0] == ids[1] + run = ids[0] + record = c.wait(run, "waiting_approval") + checkpoint = record["checkpoints"][0] + decision = {"run_id": run, "step_id": "review", "attempt": 1, "artifact_version": checkpoint["artifact_version"], "decision": "approve"} + with ThreadPoolExecutor(2) as pool: + replies = list(pool.map(lambda target: c.action(target, "decide", decision), ("a", "b"))) + assert all(r["status"] == "approved" for r in replies) + record = c.wait(run, "completed") + threads = [s["thread_id"] for s in record["steps"]] + assert c.sql("SELECT count(*) FROM runs WHERE thread_id=ANY(%s)", (threads,))[0][0] == 2 + assert c.sql("SELECT count(*) FROM workflow_controls WHERE workflow=%s", (run,))[0][0] == 1 + + +def test_shared_owner_limit_across_gateways(cluster): + from concurrent.futures import ThreadPoolExecutor + + c = cluster + runs = [] + + def launch(i): + try: + return c.launch("await pause({stepId:'p'}); return 1;", "a" if i % 2 else "b") + except AssertionError as exc: + assert "422" in str(exc) + return None + + try: + with ThreadPoolExecutor(5) as pool: + runs = [r for r in pool.map(launch, range(5)) if r] + assert len(runs) == 4 + finally: + for run in runs: + c.wait(run, "paused") + c.action("b", "cancel", {"run_id": run}) + c.wait(run, "cancelled") + + +def test_revoked_session_and_changed_configuration_block_takeover(cluster): + c = cluster + run = c.launch("await pause({stepId:'p'}); return 23;") + c.wait(run, "paused") + c.kill("a") + try: + c.wait(run, "recoverable") + epoch = c.sql("SELECT epoch FROM extension_execution_leases WHERE resource_id=%s", (run,)) + c.sql("UPDATE users SET token_version=token_version+1 RETURNING id") + c.action("b", "recover", {"run_id": run}, expected=401) + c.cookies = {} + login = c.request("b", "POST", "/api/v1/auth/login/local", data={"username": c.credentials["email"], "password": c.credentials["password"]}) + c.cookies = dict(login.cookies) + original = (c.root / "config.yaml").read_text() + config = yaml.safe_load(original) + config["models"][0]["temperature"] = 0.7 + (c.root / "config.yaml").write_text(yaml.safe_dump(config)) + try: + c.action("b", "recover", {"run_id": run}, expected=422) + assert c.sql("SELECT epoch FROM extension_execution_leases WHERE resource_id=%s", (run,)) == epoch + finally: + (c.root / "config.yaml").write_text(original) + c.recover(run) + c.resume(run) + assert c.wait(run, "completed")["result"] == 23 + finally: + c.start("a") + + +def test_sigstop_stale_artifact_writer_is_fenced(cluster): + c = cluster + c.arm("capture") + run = c.launch( + "await agent({stepId:'writer',agentRef:'lead_agent',task:'CREATE_EVIDENCE'});" + "const f=await captureArtifact({stepId:'file',sourceStepId:'writer',path:'/mnt/user-data/outputs/evidence.txt'});" + "await humanCheckpoint({stepId:'review',sourceStepId:'file',artifactVersion:f.artifactVersion}); return 31;" + ) + c.reached() + p = c.processes["a"] + os.kill(p.pid, signal.SIGSTOP) + try: + c.recover(run) + c.wait(run, "waiting_approval") + assert c.sql("SELECT epoch > 1 AND expires_at > extract(epoch from clock_timestamp()) FROM extension_execution_leases WHERE resource_id=%s", (run,))[0][0] + after = c.sql("SELECT data FROM workflows WHERE id=%s", (run,)) + artifact = c.sql("SELECT version,content FROM workflow_artifacts WHERE workflow=%s", (run,)) + (c.root / "a/release").touch() + finally: + os.kill(p.pid, signal.SIGCONT) + until(lambda: (c.root / "a/capture_error").exists()) + assert (c.root / "a/capture_error").read_text() == "LeaseLost" + assert c.sql("SELECT data FROM workflows WHERE id=%s", (run,)) == after + assert c.sql("SELECT version,content FROM workflow_artifacts WHERE workflow=%s", (run,)) == artifact + c.decide(run) + assert c.wait(run, "completed")["result"] == 31 + c.kill("a") + c.start("a") + + +def test_shared_child_limit_with_real_host_admissions(cluster): + c = cluster + reserved = c.sql("SELECT count(*) FROM workflow_slots")[0][0] + assert 0 <= reserved < 4 + runs = [] + for target in ("a", "b"): + (c.root / target / "hold_model").touch() + try: + for target in ("a", "b"): + runs.append(c.launch("return await parallel([()=>agent({stepId:'x',agentRef:'lead_agent',task:'HOLD_MODEL'}),()=>agent({stepId:'y',agentRef:'lead_agent',task:'HOLD_MODEL'})]);", target)) + until(lambda: all(len(c.get(run)["steps"]) == 2 for run in runs)) + threads = [s["thread_id"] for run in runs for s in c.get(run)["steps"]] + until(lambda: c.sql("SELECT count(*) FROM runs WHERE thread_id=ANY(%s)", (threads,))[0][0] == 4 - reserved) + assert c.sql("SELECT count(*) FROM workflow_slots")[0][0] == 4 + assert sum(s["status"] == "queued" for run in runs for s in c.get(run)["steps"]) == reserved + finally: + for target in ("a", "b"): + (c.root / target / "hold_model").unlink(missing_ok=True) + for run in runs: + assert len(c.wait(run, "completed")["steps"]) == 2 + assert c.sql("SELECT count(*) FROM workflow_slots")[0][0] == reserved + + +def test_peer_cancel_reaches_real_active_child(cluster): + c = cluster + (c.root / "a/hold_model").touch() + try: + run = c.launch("return await agent({stepId:'child',agentRef:'lead_agent',task:'HOLD_MODEL'});") + until(lambda: c.get(run)["steps"] and c.get(run)["steps"][0]["status"] == "running") + thread = c.get(run)["steps"][0]["thread_id"] + c.action("b", "cancel", {"run_id": run}) + until(lambda: c.sql("SELECT status FROM runs WHERE thread_id=%s", (thread,))[0][0] == "interrupted", 30) + # The extension has a bounded child-observation window; if the host's + # heartbeat consumes cancellation later, uncertainty must stay explicit. + until(lambda: c.get(run)["status"] in ("cancelled", "needs_reconciliation")) + assert c.sql("SELECT count(*) FROM runs WHERE thread_id=%s", (thread,))[0][0] == 1 + finally: + (c.root / "a/hold_model").unlink(missing_ok=True) + + +def test_sigkill_during_cancel_takeover_can_finish_and_delete(cluster): + from concurrent.futures import ThreadPoolExecutor + + c = cluster + run = c.launch("await pause({stepId:'p'}); return 1;") + c.wait(run, "paused") + c.kill("a") + try: + c.wait(run, "recoverable") + c.arm("cancelling", "b") + with ThreadPoolExecutor(1) as pool: + request = pool.submit(c.action, "b", "cancel", {"run_id": run}) + c.reached("b") + assert c.sql("SELECT status FROM workflows WHERE id=%s", (run,))[0][0] == "cancelling" + c.kill("b") + with pytest.raises(httpx.TransportError): + request.result(timeout=10) + c.start("a") + c.wait(run, "recoverable", "a") + # A pending cancellation must never replay the workflow body. + c.action("a", "recover", {"run_id": run}, expected=422) + assert c.action("a", "cancel", {"run_id": run})["status"] == "cancelled" + c.action("a", "delete", {"run_id": run}) + assert not c.sql("SELECT id FROM workflows WHERE id=%s", (run,)) + finally: + if "a" not in c.processes: + c.start("a") + if "b" not in c.processes: + c.start("b") diff --git a/backend/uv.lock b/backend/uv.lock index f8c086d3c9..c685c1f21d 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -966,7 +966,7 @@ extensions = [] [[package]] name = "deerflow-extension-api" -version = "0.2.8" +version = "0.2.9" source = { editable = "packages/extension-api" } [[package]] diff --git a/docs/plans/workflows-phase4.md b/docs/plans/workflows-phase4.md new file mode 100644 index 0000000000..dc0e036cb0 --- /dev/null +++ b/docs/plans/workflows-phase4.md @@ -0,0 +1,122 @@ +# Phase 4 implementation contract + +- Opt-in PostgreSQL workflow store; default SQLite/exclusive flock remains unchanged. +- Shared workflow rows/artifacts/definitions, durable control inbox, database-clock execution leases and epoch fencing. +- Generic host ExecutionFence checked in same RunStore transaction as child admission. A preflight-only lease check is insufficient. +- Lease loss stops local worker; stale owner cannot save, dispatch, approve, capture artifacts or cancel a successor's children. +- Remote browser actions revalidate current permission/config, commit exact-version decisions, and owner consumes durable controls. Duplicates are idempotent. +- Lease expiry exposes recoverable/reconciliation state. Takeover requires fresh authenticated browser capability; no credentials serialized and no autonomous execution under a reconstructed user ID. +- Cluster quotas are transactionally reserved; uncertain child effects retain capacity until reconciled, never automatically redispatched. +- Runtime/config incompatibility blocks takeover; old compatible replicas may continue their own live runs during rolling upgrade. +- Tests: opt-in deployed two-Gateway HTTP/auth/PostgreSQL/Redis/nginx acceptance, real process faults, and complementary PostgreSQL transaction/fencing regressions. + + +## Validation boundaries + +`backend/tests/test_workflow_gateway_acceptance.py` starts two real Gateways with +normal login/CSRF, shared PostgreSQL 17, Redis 7, and nginx round-robin routing. +The deterministic model replaces provider answers, not AgentRuns, admission, +LangGraph/checkpoints, storage or authorization. Test-only barriers pause real +operations before SIGKILL/SIGSTOP; no production fault hook or auth bypass exists. + +| Scenario | Deployed assertion | +| --- | --- | +| A launch; B inspection/resume/approval/cancel | Same session, real workflow reads through both proxy upstreams without affinity | +| Duplicate launch and approval | Concurrent HTTP requests to A/B produce one workflow, one decision, one child per step | +| Admission receipt lost | Actual child committed, Gateway killed before workflow receipt; recovery retains uncertain slot and never redispatches | +| Completed child, result not yet persisted | Kill before result write; survivor reads the known completed child without starting another | +| Result persisted, response not delivered | Kill after result commit; replay reuses retained output | +| Approval persisted, downstream not dispatched | Kill after decision commit; survivor preserves approval and dispatches the next step once | +| SIGSTOP old owner and restore after takeover | Actual stale host admission rejected; old workflow/artifact writes cannot change successor state | +| File approval after owner death | Modify original after capture; B reviews/approves and downstream receives original immutable bytes | +| Shared quotas | Per-owner workflow limit and cluster child reservations constrain real HTTP launches/host admissions | +| Revocation/config drift | Revoked cookie is rejected; fresh login still cannot bypass changed configuration; rejected takeover does not advance epoch | +| Rolling runtime change | Separate source trees with different worker IPC limits execute; old owner drains, incompatible survivor rejects takeover | +| Active-child cancellation | Peer request reaches actual child owner; delayed observation remains an explicit reconciliation outcome | + +The database-level suite additionally tests stale cancellation and admission while +waiting for a real PostgreSQL row lock. It complements the process faults; it is +not a claim that every transaction boundary has been exercised over HTTP. +The local deployment uses OrbStack containers and native Gateway processes with +shared local storage. This does not certify Kubernetes, arbitrary network partitions, +all published-version upgrade pairs, external business-effect rollback, or model +quality. The alternate runtime is a test source revision, not a historical release. + +### Authorization and reconciliation delivery boundary + +Lease expiry alone grants no user authority. Takeover is an explicit action using a +fresh valid browser-session capability; current permissions, configuration and exact +runtime compatibility are rechecked. No serialized credentials or reconstructed user +ID substitutes for that capability. Unattended automatic takeover needs a separate +durable delegation/authorization design and is not delivered here. + +Unknown effects retain the workflow/step evidence and child reservation. Manual +inspection can establish the business outcome, but the product currently has no +supported API/UI that records an operator attestation, audits it, and atomically +releases that reservation. That closure remains open; do not equate direct SQL, +workflow deletion, or a new request ID with a supported safe retry. Cancellation is +not rollback. Host cancellation may be observed after the extension's bounded wait, +leaving `needs_reconciliation` even when the child later becomes interrupted. + +### Defect found by deployed acceptance + +Pydantic JSON dumps converted `tool_progress.exempt_tools` from a set into a +process-dependent list order. Identical configuration on A and B could therefore +produce different durable HMACs and reject peer controls. Revision encoding now +retains Python set identity and canonicalizes sets while preserving ordered lists. +A focused regression and Gateways with different `PYTHONHASHSEED` values exercise +this fix. Existing pre-fix fingerprints may fail compatibility checks; drain them +on their original compatible runtime rather than rewriting stored fingerprints. + +## Earlier validation (before deployed acceptance) + +- Full offline backend suite with PostgreSQL enabled: 34,527 passed, 686 skipped. +- Final extension/recovery/API subset: 144 passed, including 14 cluster cases. +- Host fencing, AgentRuns, run ownership and SQL repository subset: 213 passed. +- Blocking-I/O suite: 262 passed, 1 skipped. +- Backend and extension lint/format, frontend lint/typecheck, browser fixture and + guidance checks passed. PostgreSQL 17 ran on loopback with disposable schemas. +- The final targeted runs include the owner predicate on fenced cancellation and + file-integrity validation for remote approval added during the full-suite run. + +## Deployed acceptance follow-up (2026-10-19) + +- OrbStack PostgreSQL 17 / Redis 7 / nginx, native Python 3.12 Gateways: + **14 deployed acceptance cases passed** (157 seconds). +- Final strengthened proxy/readiness and active-successor stale-artifact checks: + **1 passed each**, after the full deployed run. +- Full backend offline target with PostgreSQL regressions enabled: + **34,573 passed, 656 skipped** across four shards. The deployed suite is explicitly + opt-in and skipped in that target; its result is reported separately above. +- Blocking-I/O target: **262 passed, 1 skipped**. +- PostgreSQL coordination/fencing and AgentRuns subset: **48 passed**. +- AgentRuns and guidance checks: **38 passed**. Changed Python files passed Ruff + lint/format checks; `git diff --check` passed. +- Acceptance teardown removed its Gateway processes and disposable containers; + pytest temporary directories retain diagnostic logs. No production/user stack + was reused or modified. The standalone PostgreSQL regression server was stopped. + +## Review follow-up (2026-10-20) + +Expired `cancelling` records now project as recoverable so authenticated cancellation +can reclaim the lease and finish cleanup. Ordinary replay rejects stored cancelling, +even without a cancel inbox entry; cancel claims themselves persist cancelling so a +second crash cannot lose the cancellation intent. Cancel takeover bypasses workflow +admission caps while retaining fencing and current permission/configuration checks. +Uncertain child reservations still require reconciliation before deletion. + +Workflow get/list and database-identity reads use read-only transactions without +the global budget advisory lock. Listing joins workflow records and lease visibility +in one query/connection. Writes retain conservative global serialization; pooling +and finer write-lock granularity remain deferred for this bounded mode. + +Regressions reproduce expired cancellation with and without saturated limits and +inspection while another connection holds the budget lock. Deployed acceptance adds +SIGKILL during cancellation takeover, then cleanup and deletion through a survivor. + +Review validation: 17 PostgreSQL cluster cases passed (the three new regressions +failed before the fix); 15 real deployed acceptance cases passed. Full backend +regression: 34,576 passed, 657 skipped; blocking-I/O: 262 passed, 1 skipped. +Backend format/lint, frontend lint/typecheck, extension browser checks and diff +whitespace checks passed. The acceptance processes/containers and the dedicated +PostgreSQL regression server were cleaned up after validation. diff --git a/examples/deerflow-extension-workflows/AGENTS.md b/examples/deerflow-extension-workflows/AGENTS.md index 456e0a93f8..4dc20319b3 100644 --- a/examples/deerflow-extension-workflows/AGENTS.md +++ b/examples/deerflow-extension-workflows/AGENTS.md @@ -25,7 +25,7 @@ separately. Hash the Agent allowlist in canonical sorted order. Owner scope appl to every public read and mutation. Own persistence through supervisor publication even if the launching caller cancels. Serialize launch admission and shutdown. After artifact IO, validate call limits -and step uniqueness together with await-free reservation. Hold the process lock across service life; +and step uniqueness together with await-free reservation. In SQLite mode hold the process lock across service life; do not recover live records while another Gateway owns the file. Anchor results on the visible human input, including canonical __user rewrites; @@ -56,7 +56,16 @@ before host effects and responses before worker delivery. Include worker/supervi source and engine version in compatibility checks; Date/random are unavailable. Preserve original wait expiry and remaining execution budget. Read captured files by stored step/version after crashes, never overwrite them from mutable sources. -Phase 4 leases/fenced dispatch remain unimplemented: retain the exclusive flock. +PostgreSQL mode uses postgres.py/cluster.py: the host and extension must share +one database/schema. Database-clock leases use monotonically increasing epochs. +Every supervisor write, file capture, child admission and child cancellation is +fenced. Host admission/cancel lock the lease row in their transaction; never replace +this with a preflight check. Durable controls are immutable and consumed by the +owner. Lease loss kills only the local JS worker; never clean up successor children. +Unknown child reservations retain cluster capacity. Takeover requires a fresh +session and matching runtime/config; there is no unattended authority reattachment. +Keep the SQLite flock default. Cluster tests use DEERFLOW_TEST_WORKFLOW_POSTGRES +and disposable schemas, including real process kill and a host row-lock race. Recovery tests: backend/tests/test_workflow_recovery.py, including real process kill. AgentRuns.configuration_revision(durable=True) uses a purpose-separated HMAC @@ -64,3 +73,28 @@ under the persisted host signing secret for restart comparison. Default revision remain process-local. Rotation invalidates comparisons; no raw config or secret is returned. Recovery callers must bind a fresh authenticated session capability; never serialize handles or reconstruct authority from a stored user ID alone. + +Deployed acceptance: `backend/tests/test_workflow_gateway_acceptance.py` opts in +with DEERFLOW_WORKFLOW_ACCEPTANCE=1 and disposable Docker PostgreSQL/Redis/nginx. +The support/workflow_gateway.py entrypoint is test-only: deterministic model and +barriers around real operations, never a deployed auth/admission bypass. Use real +SIGKILL/SIGSTOP, login/CSRF, host runs and persisted assertions. Preserve uncertain +reservations; there is no supported reconciliation attestation/release API yet. + +Keep the plugin registration and execution-lease namespace in constants.py as a +single declaration. Include that module in the recovery runtime signature. + +Project expired cancelling as recoverable for cancel cleanup; never permit replay +of stored cancelling. Cancel claims persist cancelling and bypass admission caps. +Inspection/list reads use read-only, lock-free transactions and a single joined +record/lease snapshot. Keep write serialization and fencing intact; pooling remains +future work. Cover killed cancellation takeover and saturated cancellation cleanup. + +Keep the supervisor outcome private until child tasks and fenced cleanup finish. +Child saves share the workflow record; publishing terminal status early retires +the PostgreSQL lease before reservation release. Pin cancellation ordering with +a barrier-based cluster regression. + +Proven pre-admission rejection releases child capacity before retiring the lease; +unknown admission errors retain it. Share the slot-release predicate across cleanup paths. +Acceptance routing counts final Gateway ports, not nginx retry-history strings. diff --git a/examples/deerflow-extension-workflows/README.md b/examples/deerflow-extension-workflows/README.md index 7fb1c5d1a8..44039060e6 100644 --- a/examples/deerflow-extension-workflows/README.md +++ b/examples/deerflow-extension-workflows/README.md @@ -1,4 +1,4 @@ -# Script workflows (Phase 3) +# Script workflows (Phase 4) An opt-in packaged extension that runs inspectable JavaScript orchestration in the background. Scripts call full DeerFlow Agents through the public `AgentRuns` @@ -9,14 +9,14 @@ progress, child conversation/run identities, and final result. Versioned definitions, retained output snapshots, explicit same-process pause/resume, and authenticated human checkpoints support staged inspection reports. Agent success, script verdicts, tool permission approval and human output approval remain separate facts. -Compatible workflows can be recovered explicitly after a single Gateway restart. +Compatible workflows can be recovered after restart or, in PostgreSQL mode, on another Gateway after lease expiry. No arbitrary JavaScript continuation serialization or hard monetary budget is provided. ## Install and enable -Requires a POSIX Gateway (Linux/macOS), Python 3.12+, the public extension API 0.2.8, -and a **single Gateway process per storage file**. Windows execution is unsupported. -The package holds an exclusive process lock; another process using that same file +Requires a POSIX Gateway (Linux/macOS), Python 3.12+, the public extension API 0.2.9, +and a **single Gateway process per SQLite storage file**, or the shared PostgreSQL +mode described below. Windows execution is unsupported. SQLite holds an exclusive process lock; another process using that same file fails service startup rather than recovering another worker's live workflows. Use only a trusted operator-installed package. @@ -261,7 +261,7 @@ reservation/enforcement contract before advertising that guarantee. | State | Meaning | | --- | --- | -| queued / running | Supervised execution in this Gateway process | +| queued / running | Supervised execution on the current execution owner | | completed | Script and awaited child calls returned; output is unapproved task data | | failed | Script, resource limit (including wall deadline), attribution, child execution or observation failed | | cancelled | Stop requested and known active children interrupted or already terminal | @@ -398,7 +398,121 @@ and is distinct from this configuration preflight check. Phase 3 and Phase 4 share the design: stable step/attempt and idempotency identities, conditional record revisions, a durable intent/result journal, and fresh authority -are the foundations for later leases and fenced dispatch. **Phase 3 still requires -one Gateway process per workflow store.** It does not implement replica routing, -leases, automatic takeover or cluster-wide quotas; these belong to Phase 4. Keep -the exclusive process lock until those guarantees exist. +support the leases and fenced dispatch in Phase 4 below. SQLite retains the Phase 3 +single-owner contract and exclusive process lock. Both modes require explicit +session authorization for recovery; neither reconstructs credentials from records. + + +## Phase 4: shared PostgreSQL deployment + +Install the extension with its `postgres` extra. Configure every Gateway with the +same `storage_url` (a psycopg PostgreSQL DSN), `lease_seconds` (5–60, default 15), +Agent allowlist and limits. Keep `storage_path` configured for the default SQLite +mode; `storage_url` opts into PostgreSQL and disables the file lock. Existing +SQLite histories are not automatically migrated or merged. + +The DSN must resolve to the **same database and schema as the Gateway**. The host +migration installs `extension_execution_leases`; the extension creates its own +workflow, artifact, definition, control and quota tables. Startup checks database +identity and rejects differing cluster quota configurations. Use the normal +multi-instance Gateway prerequisites: `deployment.multi_instance: true`, shared +PostgreSQL and signing secret, Redis stream bridge, run ownership heartbeats, +DB run events, and consistent shared user/config/artifact storage. Credentials +stay in operator configuration; records contain only opaque configuration hashes. + +A launch through A can be inspected, approved, resumed or cancelled through B +without sticky sessions. Controls are committed durably and normally observed +within one second; a successful action acknowledges persistence, not completed +child cleanup. Exact attempt/artifact checks and immutable decision keys prevent +conflicting approvals. The owner revalidates configuration and permissions before +continuing. No model tool can approve output. + +Each execution has a database-clock lease and increasing epoch. Writes, captured +files, child admissions and child cancellation are fenced. Admission validates +the lease in the **same transaction** that creates the Agent run. A lease check +before a network call is not sufficient. Lease loss stops the local JS worker; +previously admitted Agents may have external effects and are reconciled rather +than blindly retried. An unreachable database stops progress instead of granting +local ownership. Temporary DB failure can therefore require explicit recovery. + +After expiry, any compatible Gateway exposes the workflow as recoverable. A user +signs in and chooses **Recover workflow** to supply a fresh capability. This is +session-authorized failover, not unattended failover: no raw credential or durable +user impersonation grant is stored. The same Phase 3 replay checks apply. Unknown +admissions require reconciliation; completed results and exact file versions are +reused. A pending cancellation prevents recovery from dispatching new work; cancel +the expired workflow to finish observing known children instead. + +Workflow and per-owner limits are shared; active child reservations enforce the +cluster Agent concurrency limit. An unknown child outcome keeps its reservation, +even after lease expiry. Operators must inspect the recorded thread/run before +releasing that reservation; the extension does not automatically reclaim capacity +from potentially running external effects. Reconciliation is manual in this release. +Cluster limit changes require a coordinated stop and update of the stored cluster +policy; rolling replicas cannot silently enforce different quotas. + +Rolling upgrades may continue live runs on their existing owner. Takeover requires +the exact runtime signature and matching operator/Agent configuration. An +incompatible replica rejects recovery; retain a compatible replica to drain or +cancel old workflows. This does not support arbitrary continuation conversion. + +Validation uses `DEERFLOW_TEST_WORKFLOW_POSTGRES` with a dedicated PostgreSQL +instance: `test_workflow_cluster.py` exercises independent services and process +kill, while `test_extension_execution_fence.py` checks a real PostgreSQL lock-wait +race at host admission. Agent responses are synthetic; tests make no guarantee +about undoing arbitrary business side effects. + +### Deployed acceptance and delivery boundary + +Run the opt-in HTTP/fault suite from the repository root with the backend dev +and PostgreSQL dependencies and this example's `quickjs` dependency installed: + +```sh +DEERFLOW_WORKFLOW_ACCEPTANCE=1 backend/.venv/bin/python -m pytest \ + backend/tests/test_workflow_gateway_acceptance.py -v --tb=short +``` + +Docker must be running (OrbStack works); set `DEERFLOW_TEST_DOCKER` when the CLI +is not on PATH. The suite creates disposable PostgreSQL 17, Redis 7 and nginx +containers, two real Gateway processes and a shared temporary home. It uses +normal login cookies/CSRF, actual host AgentRuns/RunManager/checkpoints and a +round-robin proxy without affinity. Only the model answers and fault barriers +are test fixtures. No paid model is called. Logs remain in pytest's temporary +directory; teardown kills only the suite's processes and removes only its named +containers/volumes. This is an opt-in deployment test, not part of offline CI. + +Coverage includes concurrent duplicate launch/approval, peer controls, shared +workflow/child limits, revoked sessions, changed configuration, SIGKILL around +admission/result/approval persistence, SIGSTOP/lease takeover/stale dispatch and +artifact writes, and immutable-file approval/consumption after source mutation +and owner death. The rolling test runs two distinct runtime source trees, with +a changed worker IPC limit, verifies live old-owner draining and rejects +incompatible takeover. It is a compatibility-guard test, not certification of +all historical releases or Kubernetes/network-partition topologies. + +**Delivered authority boundary:** lease expiry exposes recovery but never grants +permission to execute as a user. Recovery requires a fresh valid browser-session +capability and current permission/config/runtime checks. Automatic unattended +takeover requires a separately designed durable authorization/delegation contract. + +**Manual reconciliation boundary:** the current product retains stable child +identities, uncertain status and occupied child capacity. Operators must compare +the recorded conversation/run and external system effects before deciding what +happened. There is currently **no supported reconciliation API/UI to attest an +outcome, audit that decision, and release the reservation atomically**. Do not +treat direct database edits, deleting a workflow, or retrying it under a new ID +as that missing closure. This follow-up remains open. Cancellation requests do +not roll back effects; if the host consumes a cancellation after the extension's +bounded observation window, the workflow may retain `needs_reconciliation` even +when the child subsequently becomes interrupted. + +An expired owner in the middle of cancellation is exposed as recoverable for +**Cancel workflow** cleanup. Recovery rejects a persisted cancellation even if no +control-inbox row remains. Cancellation takeover bypasses workflow admission caps, +but retains lease fencing and permission/config checks; uncertain child effects +still retain reservations. A completed cancellation can be deleted normally. + +PostgreSQL workflow inspection/listing uses read-only transactions without the +cluster budget lock; listing reads records and lease visibility in one query. +Mutations still take the conservative global advisory lock, and connections are +not pooled. Those remain throughput limits of this bounded deployment mode. diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/__init__.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/__init__.py index 53a38e1afe..0350aee6c1 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/__init__.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/__init__.py @@ -12,15 +12,17 @@ from deerflow_extension_api import ( extension, ) -from .service import NAMESPACE, Workflows +from .constants import NAMESPACE +from .service import Workflows -@extension(api="0.2.8", name="workflows") +@extension(api="0.2.9", name="workflows") def install(registry, config): enabled = config.get("enabled", False) if type(enabled) is not bool: raise ValueError("enabled must be boolean") defaults = { + "lease_seconds": 15, "concurrency": 2, "global_concurrency": 4, "max_calls": 16, @@ -33,6 +35,7 @@ def install(registry, config): "memory_bytes": 16 * 1024 * 1024, } ceilings = { + "lease_seconds": 60, "concurrency": 4, "global_concurrency": 8, "max_calls": 32, @@ -44,13 +47,18 @@ def install(registry, config): "cpu_seconds": 10, "memory_bytes": 32 * 1024 * 1024, } - if set(config) - set(defaults) - {"enabled", "storage_path", "agents"}: + if set(config) - set(defaults) - {"enabled", "storage_path", "storage_url", "agents"}: raise ValueError("Unknown workflow configuration; currency budgets are not supported") + storage_url = config.get("storage_url") + if storage_url is not None and (not isinstance(storage_url, str) or not storage_url.strip()): + raise ValueError("storage_url must be a PostgreSQL connection string") resolved = {**defaults, **config} for key in defaults: value = resolved[key] if type(value) is not int or not 1 <= value <= ceilings[key]: raise ValueError(f"Invalid workflow limit {key}") + if resolved["lease_seconds"] < 5: + raise ValueError("lease_seconds must be at least 5") if resolved["memory_bytes"] < 4 * 1024 * 1024: raise ValueError("memory_bytes must be at least 4 MiB") agents = config.get("agents", []) @@ -93,7 +101,7 @@ def install(registry, config): title="Script workflows / 脚本工作流", enabled=enabled, description=( - "Bounded background script orchestration of full Agents. Versioned definitions, authenticated output checkpoints and explicit single-Gateway recovery. " + "Bounded background script orchestration of full Agents. Versioned definitions, authenticated output checkpoints and session-authorized recovery and opt-in PostgreSQL multi-Gateway control. " "Uncertain child effects require reconciliation. No hard monetary budget." ), frontend=BrowserAssets("workflows.v1", Path(__file__).parent), diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/artifacts.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/artifacts.py index 62e4d3af69..18020d7d42 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/artifacts.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/artifacts.py @@ -54,7 +54,7 @@ class Artifacts: name = text(PurePosixPath(path).name, 160) async with asyncio.timeout(10): content = await entry["runs"].read_artifact(thread_id=source["thread_id"], path=path) - info = await asyncio.to_thread(self.store.capture_artifact, owner, record["id"], step_id, name, content) + info = await asyncio.to_thread(self.store.capture_artifact, owner, record["id"], step_id, name, content, **({"execution_record": record} if self.clustered else {})) info.update( source_step_id=source["step_id"], source_run_id=source["run_id"], diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/cluster.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/cluster.py new file mode 100644 index 0000000000..e33592680e --- /dev/null +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/cluster.py @@ -0,0 +1,112 @@ +"""Leased execution and durable cross-Gateway controls, without stored authority.""" + +import asyncio +import copy +from contextlib import asynccontextmanager + +from deerflow_extension_api import ExecutionFence + +from .constants import NAMESPACE + + +class Cluster: + @staticmethod + def slot_releasable(row): + return ( + (row.get("admission_rejected") is True and row.get("run_id") is None) + or row["status"] in ("completed", "cancelled", "waiting_input") + or row.get("child_status") in ("success", "error", "interrupted", "timeout") + ) + + def dispatch_fence(self, record): + if not self.clustered: + return {} + return {"execution_fence": ExecutionFence(NAMESPACE, record["id"], self.holder_id, record["execution_epoch"])} + + @asynccontextmanager + async def cluster_slot(self, entry, row): + if not self.clustered: + yield + return + record, owner = entry["record"], entry["owner"] + while not await asyncio.to_thread(self.store.reserve_slot, owner, record, row["step_id"]): + if entry["cancel"].is_set(): + raise ValueError("Workflow stopping") + await asyncio.sleep(0.1) + try: + yield + finally: + # Unknown admission/outcome retains its cluster reservation. Losing + # an execution lease must never make a possibly live child free. + if self.slot_releasable(row) and not entry.get("lease_lost"): + await asyncio.to_thread(self.store.release_slot, owner, record, row["step_id"]) + + async def follow_lease(self, entry): + record, owner = entry["record"], entry["owner"] + try: + while record["id"] in self.entries: + async with self.write_lock: + commands = await asyncio.to_thread(self.store.heartbeat, owner, record) + changed, wake = False, False + for command in commands: + if command["kind"] == "cancel": + entry["cancel"].set() + entry["suspending"] = False + continue + row = next((r for r in record["checkpoints"] if r["step_id"] == command["step_id"]), None) + if row and row["status"] == "pending": + if row["attempt"] != command["attempt"] or row.get("artifact_version") != command["artifact_version"]: + raise ValueError("Committed checkpoint identity changed") + row.update(status=command["status"], decided_by=owner, decided_at=command["decided_at"]) + changed, wake = True, True + if changed: + record["revision"] = await asyncio.to_thread(self.store.put, owner, copy.deepcopy(record)) + if wake and entry.get("gate"): + entry["gate"].set() + await asyncio.sleep(min(1, self.config["lease_seconds"] / 3)) + except asyncio.CancelledError: + raise + except Exception: + # Fail closed on loss of contact too. The durable epoch, checked by + # every write and host admission, is the authority, not this signal. + entry["lease_lost"] = True + entry["fault"] = "Workflow execution lease lost" + entry["cancel"].set() + process = self.processes.get(record["id"]) + if process and process.returncode is None: + try: + process.kill() + except ProcessLookupError: + pass + + async def remote_command(self, owner, record, context, command): + entry = self.new_entry(owner, record, context.agent_runs) + await self.check_configuration(entry) + key = "cancel" if command["kind"] == "cancel" else command["step_id"] + await asyncio.to_thread(self.store.command, owner, record["id"], key, command, record.get("revision", 0)) + + async def cancel_recovered(self, owner, run_id, context): + record = await asyncio.to_thread(self.store.get, owner, run_id) + await self.check_configuration(self.new_entry(owner, record, context.agent_runs)) + record = await asyncio.to_thread(self.store.claim, owner, run_id, for_cancel=True) + entry = self.new_entry(owner, record, context.agent_runs) + record["status"] = "cancelling" + await self.save(owner, record) + self.entries[run_id] = entry + guardian = asyncio.create_task(self.follow_lease(entry)) + try: + for row in record["steps"]: + if entry.get("lease_lost"): + raise ValueError("Workflow execution lease lost") + if row.get("run_id") and row["status"] != "completed": + await self.cancel_child(context.agent_runs, row, record=record) + if self.slot_releasable(row): + await asyncio.to_thread(self.store.release_slot, owner, record, row["step_id"]) + uncertain = any(r["status"] not in ("completed", "queued", "cancelled") for r in record["steps"]) + record.update(status="needs_reconciliation" if uncertain else "cancelled", error="Inspect uncertain child effects" if uncertain else None) + await self.save(owner, record) + return {"run_id": run_id, "status": record["status"]} + finally: + guardian.cancel() + await asyncio.gather(guardian, return_exceptions=True) + self.entries.pop(run_id, None) diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/constants.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/constants.py new file mode 100644 index 0000000000..064894d8ca --- /dev/null +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/constants.py @@ -0,0 +1,3 @@ +"""Shared workflow identity for plugin registration and execution fencing.""" + +NAMESPACE = "community.workflows" diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/lifecycle.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/lifecycle.py index cbd5ee365d..87fa2731a2 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/lifecycle.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/lifecycle.py @@ -179,6 +179,8 @@ class Lifecycle: retained = await asyncio.to_thread(self.store.get, owner, run_id) entry = self.entries.get(run_id) record = entry["record"] if entry else retained + if self.clustered: + await self.check_configuration(self.new_entry(owner, record, context.agent_runs)) if human: row = next( (r for r in record.get("checkpoints", []) if r["step_id"] == payload["step_id"]), @@ -210,6 +212,12 @@ class Lifecycle: if row["status"] == "resumed": return {"run_id": run_id, "status": "resumed"} status = "resumed" + if self.clustered and record["status"] == ("waiting_approval" if human else "paused"): + if human and row.get("artifact_kind") == "file": + await asyncio.to_thread(self.store.artifact, owner, run_id, row["artifact_version"]) + command = {"kind": "checkpoint", "step_id": row["step_id"], "attempt": row["attempt"], "artifact_version": row.get("artifact_version"), "status": status} + await self.remote_command(owner, record, context, command) + return {"run_id": run_id, "status": status} if not entry or record["status"] != ("waiting_approval" if human else "paused") or entry["cancel"].is_set(): raise ValueError("Live checkpoint unavailable; recover the workflow first after restart") # Revalidate both the retained capability and this caller's live grant. diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/postgres.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/postgres.py new file mode 100644 index 0000000000..b74cdd599a --- /dev/null +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/postgres.py @@ -0,0 +1,247 @@ +"""Shared authoritative workflow store; all decisions use the PostgreSQL clock. + +The host migration owns extension_execution_leases. The workflow tables remain +extension-owned. Host run admission locks the same lease row in its transaction. +""" + +import json +from contextlib import closing + +from .constants import NAMESPACE +from .store import TERMINAL, Store + +LOCK = 723461108 + + +class LeaseLost(ValueError): + """An execution generation no longer has authority to mutate state.""" + + +class Connection: + def __init__(self, raw): + self.raw = raw + + def execute(self, sql, args=()): + if sql == "BEGIN IMMEDIATE": + return self.raw.execute("SELECT 1") + return self.raw.execute(sql.replace("?", "%s"), args) + + def __enter__(self): + return self + + def __exit__(self, kind, value, tb): + self.raw.rollback() if kind else self.raw.commit() + + def close(self): + self.raw.close() + + +class PostgresStore(Store): + def __init__(self, dsn, holder, config): + self.dsn, self.holder, self.config = dsn, holder, config + with closing(self.connect()) as db, db: + # Fail closed until the host schema (including the admission fence) + # is installed in this same database/search_path. + db.execute("SELECT 1 FROM extension_execution_leases LIMIT 0") + db.execute( + "CREATE TABLE IF NOT EXISTS workflows (id TEXT PRIMARY KEY, owner TEXT NOT NULL, request_id TEXT NOT NULL, fingerprint TEXT NOT NULL, status TEXT NOT NULL, data TEXT NOT NULL, " + "revision BIGINT NOT NULL DEFAULT 0, rowid BIGINT GENERATED ALWAYS AS IDENTITY, UNIQUE(owner,request_id))" + ) + db.execute( + "CREATE TABLE IF NOT EXISTS workflow_artifacts (owner TEXT NOT NULL, workflow TEXT NOT NULL, step_id TEXT NOT NULL, version TEXT NOT NULL, info TEXT NOT NULL, content BYTEA NOT NULL, " + "PRIMARY KEY(owner,workflow,step_id), UNIQUE(owner,workflow,version))" + ) + db.execute("CREATE OR REPLACE FUNCTION workflow_artifact_immutable() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'Artifact versions cannot be overwritten'; END $$") + db.execute("DROP TRIGGER IF EXISTS workflow_artifacts_immutable ON workflow_artifacts") + db.execute("CREATE TRIGGER workflow_artifacts_immutable BEFORE UPDATE ON workflow_artifacts FOR EACH ROW EXECUTE FUNCTION workflow_artifact_immutable()") + db.execute("CREATE TABLE IF NOT EXISTS workflow_definitions (owner TEXT NOT NULL, name TEXT NOT NULL, version TEXT NOT NULL, script TEXT NOT NULL, rowid BIGINT GENERATED ALWAYS AS IDENTITY, PRIMARY KEY(owner,name,version))") + db.execute("CREATE TABLE IF NOT EXISTS workflow_controls (workflow TEXT NOT NULL, key TEXT NOT NULL, data TEXT NOT NULL, PRIMARY KEY(workflow,key))") + db.execute("CREATE TABLE IF NOT EXISTS workflow_cluster_policy (id INTEGER PRIMARY KEY, policy TEXT NOT NULL)") + policy = json.dumps({k: config[k] for k in ("global_concurrency", "max_workflows", "max_workflows_per_owner", "lease_seconds")}, sort_keys=True) + db.execute("INSERT INTO workflow_cluster_policy VALUES (1,?) ON CONFLICT DO NOTHING", (policy,)) + if db.execute("SELECT policy FROM workflow_cluster_policy WHERE id=1").fetchone()[0] != policy: + raise ValueError("Workflow cluster limits differ; coordinate the operator configuration before startup") + db.execute("CREATE TABLE IF NOT EXISTS workflow_slots (workflow TEXT NOT NULL, step TEXT NOT NULL, PRIMARY KEY(workflow,step))") + + def connect(self, *, readonly=False): + import psycopg + + raw = psycopg.connect(self.dsn, connect_timeout=5, client_encoding="utf8") + try: + if readonly: + raw.execute("SET TRANSACTION READ ONLY") + raw.execute("SET LOCAL lock_timeout = '5s'") + raw.execute("SET LOCAL statement_timeout = '10s'") + # Serializes budget counts and mutations across all replicas. No + # host/Agent IO happens in this extension transaction. + if not readonly: + raw.execute("SELECT pg_advisory_xact_lock(%s)", (LOCK,)) + return Connection(raw) + except BaseException: + raw.close() + raise + + def _lease(self, db, owner, record): + row = db.execute("SELECT holder_id,epoch,expires_at FROM extension_execution_leases WHERE namespace=? AND resource_id=? AND user_id=? FOR UPDATE", (NAMESPACE, record["id"], owner)).fetchone() + now = float(db.execute("SELECT extract(epoch from clock_timestamp())").fetchone()[0]) + if not row or row[0] != self.holder or row[1] != record.get("execution_epoch") or row[2] <= now: + raise LeaseLost("Workflow execution lease lost") + return now + + def _capacity(self, db, owner, exclude=""): + active = db.execute("SELECT owner FROM workflows WHERE status NOT IN ('completed','failed','cancelled','waiting_input','interrupted','needs_reconciliation','recoverable') AND id<>?", (exclude,)).fetchall() + if len(active) >= self.config["max_workflows"] or sum(r[0] == owner for r in active) >= self.config["max_workflows_per_owner"]: + raise ValueError("Active workflow limit") + + def _claim(self, db, owner, record): + row = db.execute("SELECT epoch,expires_at FROM extension_execution_leases WHERE namespace=? AND resource_id=? FOR UPDATE", (NAMESPACE, record["id"])).fetchone() + now = float(db.execute("SELECT extract(epoch from clock_timestamp())").fetchone()[0]) + if row and row[1] > now: + raise ValueError("Workflow owned by a live Gateway") + epoch = row[0] + 1 if row else 1 + db.execute( + "INSERT INTO extension_execution_leases(namespace,resource_id,user_id,holder_id,epoch,expires_at) VALUES (?,?,?,?,?,?) ON CONFLICT(namespace,resource_id) DO UPDATE SET " + "holder_id=excluded.holder_id,epoch=excluded.epoch,expires_at=excluded.expires_at", + (NAMESPACE, record["id"], owner, self.holder, epoch, now + self.config["lease_seconds"]), + ) + record["execution_epoch"] = epoch + + def create(self, owner, record): + with closing(self.connect()) as db, db: + old = db.execute("SELECT fingerprint,data FROM workflows WHERE owner=? AND request_id=?", (owner, record["request_id"])).fetchone() + if old: + if old[0] != record["fingerprint"]: + raise ValueError("request_id already used with different script or arguments") + return json.loads(old[1]), False + self._capacity(db, owner) + if db.execute("SELECT count(*) FROM workflows WHERE owner=?", (owner,)).fetchone()[0] >= 100: + raise ValueError("Retained workflow limit (100); delete finished records") + self._claim(db, owner, record) + db.execute("INSERT INTO workflows(id,owner,request_id,fingerprint,status,data) VALUES (?,?,?,?,?,?)", (record["id"], owner, record["request_id"], record["fingerprint"], record["status"], json.dumps(record))) + return record, True + + def claim(self, owner, run_id, *, for_cancel=False): + with closing(self.connect()) as db, db: + row = db.execute("SELECT data FROM workflows WHERE owner=? AND id=?", (owner, run_id)).fetchone() + if not row: + raise ValueError("Workflow unavailable") + record = json.loads(row[0]) + if record["status"] in TERMINAL - {"recoverable"}: + raise ValueError("Workflow is not recoverable") + if not for_cancel and (record["status"] == "cancelling" or db.execute("SELECT 1 FROM workflow_controls WHERE workflow=? AND key='cancel'", (run_id,)).fetchone()): + raise ValueError("Cancellation pending; cancel this workflow instead of recovering it") + if not for_cancel: + self._capacity(db, owner, run_id) + self._claim(db, owner, record) + record.update(status="cancelling" if for_cancel else "queued", revision=record.get("revision", 0) + 1) + db.execute("UPDATE workflows SET status=?,data=?,revision=? WHERE id=?", (record["status"], json.dumps(record), record["revision"], run_id)) + return record + + def put(self, owner, record): + revision = record.get("revision", 0) + updated = {**record, "revision": revision + 1} + raw = json.dumps(updated) + if len(raw.encode()) > 8 * 1024 * 1024: + raise ValueError("Workflow recovery journal limit (8 MiB)") + with closing(self.connect()) as db, db: + self._lease(db, owner, record) + if db.execute("UPDATE workflows SET status=?,data=?,revision=revision+1 WHERE owner=? AND id=? AND revision=?", (record["status"], raw, owner, record["id"], revision)).rowcount != 1: + raise ValueError("Stale workflow revision") + if record["status"] in TERMINAL: + db.execute("UPDATE extension_execution_leases SET expires_at=0 WHERE namespace=? AND resource_id=?", (NAMESPACE, record["id"])) + record["revision"] = revision + 1 + return revision + 1 + + def heartbeat(self, owner, record): + with closing(self.connect()) as db, db: + now = self._lease(db, owner, record) + db.execute("UPDATE extension_execution_leases SET expires_at=? WHERE namespace=? AND resource_id=?", (now + self.config["lease_seconds"], NAMESPACE, record["id"])) + return [json.loads(row[0]) for row in db.execute("SELECT data FROM workflow_controls WHERE workflow=?", (record["id"],))] + + def command(self, owner, run_id, key, command, revision): + with closing(self.connect()) as db, db: + row = db.execute("SELECT revision,data FROM workflows WHERE owner=? AND id=?", (owner, run_id)).fetchone() + if not row or row[0] != revision: + raise ValueError("Workflow changed; reload and retry") + record = json.loads(row[1]) + now = float(db.execute("SELECT extract(epoch from clock_timestamp())").fetchone()[0]) + if command["kind"] != "cancel" and (record.get("pause_expires_at", 0) <= now or record["status"] not in ("paused", "waiting_approval")): + raise ValueError("Checkpoint expired or unavailable") + old = db.execute("SELECT data FROM workflow_controls WHERE workflow=? AND key=?", (run_id, key)).fetchone() + raw = json.dumps(command, sort_keys=True) + if old and {k: v for k, v in json.loads(old[0]).items() if k != "decided_at"} != command: + raise ValueError("Checkpoint already decided") + command = {**command, "decided_at": float(db.execute("SELECT extract(epoch from clock_timestamp())").fetchone()[0])} + raw = json.dumps(command, sort_keys=True) + db.execute("INSERT INTO workflow_controls VALUES (?,?,?) ON CONFLICT DO NOTHING", (run_id, key, raw)) + + def reserve_slot(self, owner, record, step): + with closing(self.connect()) as db, db: + self._lease(db, owner, record) + if db.execute("SELECT 1 FROM workflow_slots WHERE workflow=? AND step=?", (record["id"], step)).fetchone(): + return True + if db.execute("SELECT count(*) FROM workflow_slots").fetchone()[0] >= self.config["global_concurrency"]: + return False + db.execute("INSERT INTO workflow_slots VALUES (?,?)", (record["id"], step)) + return True + + def release_slot(self, owner, record, step): + with closing(self.connect()) as db, db: + self._lease(db, owner, record) + db.execute("DELETE FROM workflow_slots WHERE workflow=? AND step=?", (record["id"], step)) + + def save_definition(self, owner, name, version, script): + # Explicit columns: PostgreSQL also stores an ordering identity. + with closing(self.connect()) as db, db: + existing = db.execute("SELECT 1 FROM workflow_definitions WHERE owner=? AND name=? AND version=?", (owner, name, version)).fetchone() + if not existing and db.execute("SELECT count(*) FROM workflow_definitions WHERE owner=?", (owner,)).fetchone()[0] >= 100: + raise ValueError("Definition version limit (100)") + db.execute("INSERT INTO workflow_definitions(owner,name,version,script) VALUES (?,?,?,?) ON CONFLICT DO NOTHING", (owner, name, version, script)) + return {"name": name, "version": version} + + @staticmethod + def _project(row): + record = json.loads(row[0]) + if record["status"] not in TERMINAL and not row[1]: + record["status"] = "recoverable" + record["error"] = "Execution owner unavailable; sign in to recover or cancel pending cleanup" + return record + + def get(self, owner, run_id): + with closing(self.connect(readonly=True)) as db: + row = db.execute( + "SELECT w.data,l.expires_at > extract(epoch from clock_timestamp()) FROM workflows w LEFT JOIN extension_execution_leases l " + "ON l.namespace=? AND l.resource_id=w.id AND l.user_id=w.owner WHERE w.owner=? AND w.id=?", + (NAMESPACE, owner, run_id), + ).fetchone() + if not row: + raise ValueError("Workflow unavailable") + return self._project(row) + + def list(self, owner): + with closing(self.connect(readonly=True)) as db: + rows = db.execute( + "SELECT w.data,l.expires_at > extract(epoch from clock_timestamp()) FROM workflows w LEFT JOIN extension_execution_leases l " + "ON l.namespace=? AND l.resource_id=w.id AND l.user_id=w.owner WHERE w.owner=? ORDER BY w.rowid DESC LIMIT 20", + (NAMESPACE, owner), + ).fetchall() + return [self._project(row) for row in rows] + + def delete(self, owner, run_id): + with closing(self.connect()) as db, db: + row = db.execute("SELECT status FROM workflows WHERE owner=? AND id=?", (owner, run_id)).fetchone() + if not row or row[0] not in TERMINAL: + raise ValueError("Only finished workflow records can be deleted") + lease = db.execute("SELECT expires_at > extract(epoch from clock_timestamp()) FROM extension_execution_leases WHERE namespace=? AND resource_id=? FOR UPDATE", (NAMESPACE, run_id)).fetchone() + if lease and lease[0]: + raise ValueError("Workflow still has an execution owner") + if db.execute("SELECT 1 FROM workflow_slots WHERE workflow=?", (run_id,)).fetchone(): + raise ValueError("Reconcile retained child reservations before deleting") + db.execute("DELETE FROM workflow_artifacts WHERE owner=? AND workflow=?", (owner, run_id)) + db.execute("DELETE FROM workflow_controls WHERE workflow=?", (run_id,)) + db.execute("DELETE FROM workflows WHERE owner=? AND id=?", (owner, run_id)) + # Keep the lease generation tombstone: deletion never resets fencing. + + def database_identity(self): + with closing(self.connect(readonly=True)) as db: + return tuple(db.execute("SELECT current_database(),current_schema(),inet_server_addr()::text,inet_server_port()").fetchone()) diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/recovery.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/recovery.py index 4b6af5138b..e1f55cb10d 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/recovery.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/recovery.py @@ -1,7 +1,7 @@ """Versioned bridge journal replay; authority is reacquired from a live session. -Phase 3 retains the exclusive store owner. Conditional revisions and durable -bridge intents are the foundation for a future leased, fenced cluster owner. +SQLite retains its exclusive store owner. PostgreSQL uses a shared execution +lease and fenced admission; both replay the same committed bridge journal. """ import asyncio @@ -43,6 +43,9 @@ class Recovery: raise ValueError("Active workflow limit") entry = self.new_entry(owner, record, context.agent_runs) await self.check_configuration(entry) + if self.clustered: + record = await asyncio.to_thread(self.store.claim, owner, run_id) + entry = self.new_entry(owner, record, context.agent_runs) # A committed admission intent with no receipt is never dispatched # again. An operator must reconcile its external effects first. uncertain = [s for s in record["steps"] if s["status"] not in ("queued", "completed", "running") or (s["status"] == "running" and not s.get("run_id"))] diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py index 2ceb37c5cb..820018dcb1 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py @@ -16,11 +16,13 @@ from pathlib import Path from deerflow_extension_api import AgentRunError from .artifacts import Artifacts +from .cluster import Cluster +from .constants import NAMESPACE as NAMESPACE from .lifecycle import Lifecycle +from .postgres import LeaseLost from .recovery import RECOVERY_VERSION, Recovery from .store import TERMINAL, Store -NAMESPACE = "community.workflows" FRAME = 48 * 1024 TURN_LIMIT = 10000 AGENT_TASK_BYTES = 16000 @@ -97,9 +99,12 @@ def project(record, detail=True): return result -class Workflows(Recovery, Lifecycle, Artifacts): +class Workflows(Cluster, Recovery, Lifecycle, Artifacts): def __init__(self, config): self.config = config + self.clustered = bool(config.get("storage_url")) + self.holder_id = uuid.uuid4().hex + self.heartbeats = {} self.store = None self.available = False self.workers = {} @@ -118,31 +123,51 @@ class Workflows(Recovery, Lifecycle, Artifacts): path = Path(self.config["storage_path"]) path.parent.mkdir(parents=True, exist_ok=True) - lock = path.with_suffix(path.suffix + ".lock").open("a+b") + lock = None if self.clustered else path.with_suffix(path.suffix + ".lock").open("a+b") try: - fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - store = Store(path) - store.interrupt_unfinished() + if self.clustered: + from .postgres import PostgresStore + + store = PostgresStore(self.config["storage_url"], self.holder_id, self.config) + else: + fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + store = Store(path) + store.interrupt_unfinished() # Include worker code and engine version in the conservative # replay compatibility guard; upgrades never silently replay. from importlib.metadata import version signature = hashlib.sha256() for name in ( + "constants.py", "worker.py", "service.py", "recovery.py", "lifecycle.py", "artifacts.py", + "cluster.py", + "postgres.py", ): signature.update(Path(__file__).with_name(name).read_bytes()) signature.update(version("quickjs").encode()) return lock, store, signature.hexdigest() except BaseException: - lock.close() + if lock: + lock.close() raise self.lockfile, self.store, self.runtime_version = await asyncio.to_thread(initialize) + if self.clustered and deps is not None: + from sqlalchemy import text as sql_text + + if deps.session_factory is None: + raise ValueError("Cluster workflows require the host PostgreSQL session factory") + async with deps.session_factory() as session: + if session.get_bind().dialect.name != "postgresql": + raise ValueError("Cluster workflows require host PostgreSQL persistence") + host_identity = tuple((await session.execute(sql_text("SELECT current_database(),current_schema(),inet_server_addr()::text,inet_server_port()"))).one()) + if host_identity != await asyncio.to_thread(self.store.database_identity): + raise ValueError("Workflow and host must share the same PostgreSQL database and schema") self.available = True async def stop(self): @@ -158,6 +183,12 @@ class Workflows(Recovery, Lifecycle, Artifacts): await asyncio.gather(*list(self.admissions), return_exceptions=True) if workers: await asyncio.gather(*workers, return_exceptions=True) + if self.heartbeats: + heartbeats = list(self.heartbeats.values()) + for heartbeat in heartbeats: + heartbeat.cancel() + await asyncio.gather(*heartbeats, return_exceptions=True) + self.heartbeats.clear() if self.lockfile: await asyncio.to_thread(self.lockfile.close) self.lockfile = None @@ -190,7 +221,14 @@ class Workflows(Recovery, Lifecycle, Artifacts): async def save_drained(self, owner, record): async with self.write_lock: - record["revision"] = await asyncio.to_thread(self.store.put, owner, copy.deepcopy(record)) + try: + record["revision"] = await asyncio.to_thread(self.store.put, owner, copy.deepcopy(record)) + except LeaseLost: + entry = self.entries.get(record["id"]) + if entry: + entry["lease_lost"] = True + entry["cancel"].set() + raise async def launch(self, payload, context): owner = self.owner(context) @@ -293,7 +331,8 @@ class Workflows(Recovery, Lifecycle, Artifacts): if len(self.entries) >= self.config["max_workflows"] or len(current) >= self.config["max_workflows_per_owner"]: raise ValueError("Active workflow limit") record, created = await asyncio.to_thread(self.store.create, owner, record) - self.publish(self.new_entry(owner, record, context.agent_runs)) + if created: + self.publish(self.new_entry(owner, record, context.agent_runs)) return {"run_id": record["id"], "status": record["status"]} def new_entry(self, owner, record, runs): @@ -310,11 +349,16 @@ class Workflows(Recovery, Lifecycle, Artifacts): def publish(self, entry): run_id = entry["record"]["id"] self.entries[run_id] = entry + if self.clustered: + self.heartbeats[run_id] = asyncio.create_task(self.follow_lease(entry)) task = asyncio.create_task(self.execute(entry)) self.workers[run_id] = task task.add_done_callback(lambda t: self.worker_finished(run_id, t)) def worker_finished(self, run_id, task): + heartbeat = self.heartbeats.pop(run_id, None) + if heartbeat: + heartbeat.cancel() self.workers.pop(run_id, None) self.entries.pop(run_id, None) if not task.cancelled() and task.exception() is not None: @@ -347,6 +391,11 @@ class Workflows(Recovery, Lifecycle, Artifacts): run_id = text(payload.get("run_id"), 64) record = await asyncio.to_thread(self.store.get, owner, run_id) entry = self.entries.get(run_id) + if self.clustered and record["status"] not in TERMINAL: + await self.remote_command(owner, record, context, {"kind": "cancel"}) + return {"run_id": run_id, "status": "cancelling"} + if self.clustered and record["status"] == "recoverable": + return await self.cancel_recovered(owner, run_id, context) if entry: entry["cancel"].set() entry["suspending"] = False @@ -454,7 +503,7 @@ class Workflows(Recovery, Lifecycle, Artifacts): } record["steps"].append(row) await self.save(owner, record) - async with entry["slots"], self.global_slots: + async with entry["slots"], self.global_slots, self.cluster_slot(entry, row): if entry["cancel"].is_set() or entry["fault"]: row["status"] = "cancelled" raise ValueError("Workflow stopping") @@ -464,6 +513,8 @@ class Workflows(Recovery, Lifecycle, Artifacts): except Exception: entry["fault"] = "Agent configuration changed or unavailable" row["status"] = "failed" + if row.get("run_id") is None: + row["admission_rejected"] = True raise if row["status"] == "queued": row["status"] = "creating_thread" @@ -485,6 +536,7 @@ class Workflows(Recovery, Lifecycle, Artifacts): thread_id=thread_id, input=input, idempotency_key=row["idempotency_key"], + **self.dispatch_fence(record), ) row.update( run_id=run.run_id, @@ -496,7 +548,8 @@ class Workflows(Recovery, Lifecycle, Artifacts): run = await runs.get(thread_id=thread_id, run_id=row["run_id"]) while True: if entry["cancel"].is_set() or entry["fault"]: - await self.cancel_child(runs, row) + if not entry.get("lease_lost"): + await self.cancel_child(runs, row, **({"record": record} if self.clustered else {})) raise ValueError("Workflow stopping") async with asyncio.timeout(5): run = await runs.get(thread_id=thread_id, run_id=run.run_id) @@ -570,6 +623,8 @@ class Workflows(Recovery, Lifecycle, Artifacts): response["error"] = entry["fault"] except AgentRunError as exc: if row: + if row.get("run_id") is None and exc.status_code in (400, 403, 404, 409, 422): + row["admission_rejected"] = True row["status"] = "failed" if exc.status_code in (400, 403, 404, 409, 422) else "needs_reconciliation" response["error"] = f"AgentRuns rejected operation ({exc.status_code})" entry["fault"] = response["error"] @@ -600,12 +655,12 @@ class Workflows(Recovery, Lifecycle, Artifacts): "status": "success", } - async def cancel_child(self, runs, row): + async def cancel_child(self, runs, row, record=None): try: async with asyncio.timeout(5): run = await runs.get(thread_id=row["thread_id"], run_id=row["run_id"]) if run.status in ("pending", "running"): - await runs.cancel(thread_id=row["thread_id"], run_id=row["run_id"]) + await runs.cancel(thread_id=row["thread_id"], run_id=row["run_id"], **(self.dispatch_fence(record) if record is not None else {})) run = await runs.wait(thread_id=row["thread_id"], run_id=row["run_id"], timeout=4) row["child_status"] = run.status if row["status"] == "running": @@ -617,6 +672,7 @@ class Workflows(Recovery, Lifecycle, Artifacts): owner, record, runs = entry["owner"], entry["record"], entry["runs"] process = None workspace = None + final_state = {} try: if record.get("budget_deadline") is not None: record["remaining_seconds"] = max(0, record["budget_deadline"] - time.time()) @@ -717,7 +773,7 @@ class Workflows(Recovery, Lifecycle, Artifacts): message = outcome.get("message") raise ValueError(message[:240] if isinstance(message, str) else "Script failed or exceeded limits") encoded(outcome["value"], 12000) - record.update(status="completed", result=outcome["value"]) + final_state.update(status="completed", result=outcome["value"]) break # Poll cancel/fault/worker exit while a response is pending. An idle worker can only progress # on a response, so keep waiting here instead of sending empty frames (turns). @@ -741,15 +797,15 @@ class Workflows(Recovery, Lifecycle, Artifacts): except TimeoutError: # Keep this outside asyncio.timeout: its own cancellation must first # leave the context manager so it can be translated to TimeoutError. - record.update( + final_state.update( status="failed", error="Checkpoint wait deadline exceeded" if record["status"] in ("paused", "waiting_approval") else "Execution deadline exceeded", ) except asyncio.CancelledError: entry["cancel"].set() - record.update(status="interrupted", error="Host interrupted; child cleanup required") + final_state.update(status="interrupted", error="Host interrupted; child cleanup required") except Exception as exc: - record.update( + final_state.update( status="cancelled" if entry["cancel"].is_set() else "waiting_input" if entry["fault"] == "waiting_input" else "failed", error=str(exc)[:240], ) @@ -768,9 +824,18 @@ class Workflows(Recovery, Lifecycle, Artifacts): # Admission tasks run to a bounded completion; no lost response on workflow cancel. if entry["tasks"]: await asyncio.gather(*entry["tasks"], return_exceptions=True) - cleanup = [self.cancel_child(runs, row) for row in record["steps"] if row.get("run_id") and row["status"] not in ("completed", "waiting_input", "cancelled")] + if entry.get("lease_lost"): + return + cleanup = [self.cancel_child(runs, row, **({"record": record} if self.clustered else {})) for row in record["steps"] if row.get("run_id") and row["status"] not in ("completed", "waiting_input", "cancelled")] if cleanup: await asyncio.gather(*cleanup) + if self.clustered: + for row in record["steps"]: + if self.slot_releasable(row): + await asyncio.to_thread(self.store.release_slot, owner, record, row["step_id"]) + # Child saves must not publish a terminal workflow and retire its + # lease before all fenced cleanup and reservation releases finish. + record.update(final_state) if any(s["status"] == "needs_reconciliation" for s in record["steps"]): record.update( status="needs_reconciliation", diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/static/index.mjs b/examples/deerflow-extension-workflows/deerflow_extension_workflows/static/index.mjs index 17b4cf05dc..450bdee5dd 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/static/index.mjs +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/static/index.mjs @@ -48,7 +48,7 @@ function mount(root, context) { const form = node('form'); const submit = node('button',t('Launch workflow','启动工作流'),{type:'submit'}); form.append(node('label',t('Inspectable JavaScript script','可检查的 JavaScript 脚本')),script,node('label',t('Arguments (JSON object)','参数（JSON 对象）')),args,submit); - panel.append(node('h2',t('Script workflows','脚本工作流')),node('p',t('Versioned workflows with authenticated output review. After restart, sign in and recover compatible runs. Single Gateway only; uncertain child effects require reconciliation; no hard monetary budget. File approval covers a verified immutable snapshot; linked files require explicit capture.','版本化工作流支持登录用户成果确认。重启后请登录并恢复兼容的工作流。仅支持单 Gateway；不确定的子任务效果需要核对；无金额硬上限。文件确认针对经过哈希验证的不可覆盖快照；链接文件需显式捕获。')),notice,library,form,navigation,detail); + panel.append(node('h2',t('Script workflows','脚本工作流')),node('p',t('Versioned workflows with authenticated output review. After restart, sign in and recover compatible runs. Multi-Gateway control requires the shared PostgreSQL mode; takeover requires sign-in; uncertain child effects require reconciliation; no hard monetary budget. File approval covers a verified immutable snapshot; linked files require explicit capture.','版本化工作流支持登录用户成果确认。重启后请登录并恢复兼容的工作流。多 Gateway 控制需启用共享 PostgreSQL 模式；接管需要登录；不确定的子任务效果需要核对；无金额硬上限。文件确认针对经过哈希验证的不可覆盖快照；链接文件需显式捕获。')),notice,library,form,navigation,detail); root.append(sheet,panel); const alive = version => !disposed && !context.signal?.aborted && version === generation; async function call(action,payload) { diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/store.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/store.py index f1c51892df..66111d370b 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/store.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/store.py @@ -36,7 +36,7 @@ class Store: db.execute("CREATE TRIGGER IF NOT EXISTS workflow_artifacts_immutable BEFORE UPDATE ON workflow_artifacts BEGIN SELECT RAISE(ABORT, 'Artifact versions cannot be overwritten'); END") db.execute("CREATE TABLE IF NOT EXISTS workflow_definitions (owner TEXT NOT NULL, name TEXT NOT NULL, version TEXT NOT NULL, script TEXT NOT NULL, PRIMARY KEY(owner,name,version))") - def capture_artifact(self, owner, workflow, step_id, name, content): + def capture_artifact(self, owner, workflow, step_id, name, content, *, execution_record=None): if not isinstance(content, bytes) or len(content) > 2 * 1024 * 1024: raise ValueError("Artifact limit is 2 MiB") digest = hashlib.sha256(content).hexdigest() @@ -54,6 +54,8 @@ class Store: } with closing(self.connect()) as db, db: db.execute("BEGIN IMMEDIATE") + if execution_record is not None: + self._lease(db, owner, execution_record) old = db.execute( "SELECT info FROM workflow_artifacts WHERE owner=? AND workflow=? AND step_id=?", (owner, workflow, step_id), diff --git a/examples/deerflow-extension-workflows/pyproject.toml b/examples/deerflow-extension-workflows/pyproject.toml index fd3295e855..b131fd9845 100644 --- a/examples/deerflow-extension-workflows/pyproject.toml +++ b/examples/deerflow-extension-workflows/pyproject.toml @@ -1,10 +1,10 @@ [project] name = "deerflow-extension-workflows" -version = "0.3.0" +version = "0.4.0" description = "Opt-in constrained script orchestration of public DeerFlow AgentRuns" requires-python = ">=3.12" license = "MIT" -dependencies = ["deerflow-extension-api>=0.2.8,<0.3", "quickjs==1.19.4"] +dependencies = ["deerflow-extension-api>=0.2.9,<0.3", "quickjs==1.19.4"] [project.entry-points."deerflow.extensions"] workflows = "deerflow_extension_workflows:install" @@ -15,3 +15,6 @@ build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["deerflow_extension_workflows"] + +[project.optional-dependencies] +postgres = ["psycopg[binary]>=3.2,<4", "sqlalchemy>=2,<3"]