From 5650a80b0e292ffc5ab1722864654f67d11dc205 Mon Sep 17 00:00:00 2001 From: hetaoBackend Date: Fri, 7 Aug 2026 04:57:40 +0800 Subject: [PATCH] fix: bind remote deletion recovery to sandbox incarnations --- README.md | 7 +- backend/AGENTS.md | 2 +- backend/docs/FILE_UPLOAD.md | 4 +- .../aio_sandbox/aio_sandbox_provider.py | 51 ++-- .../deerflow/community/aio_sandbox/backend.py | 36 ++- .../community/aio_sandbox/local_backend.py | 49 +++- .../community/aio_sandbox/remote_backend.py | 35 ++- .../deerflow/sandbox/sandbox_provider.py | 37 ++- .../harness/deerflow/uploads/layout.py | 1 + .../harness/deerflow/uploads/manager.py | 43 +++- .../harness/deerflow/uploads/sandbox_sync.py | 148 ++++++++++- .../tests/test_aio_sandbox_local_backend.py | 36 +++ backend/tests/test_aio_sandbox_provider.py | 152 ++++++++++- backend/tests/test_client.py | 33 ++- .../test_provisioner_request_threading.py | 111 ++++++++ backend/tests/test_remote_sandbox_backend.py | 31 ++- backend/tests/test_upload_sandbox_sync.py | 240 +++++++++++++++++- backend/tests/test_uploads_router.py | 64 +++-- docker/provisioner/README.md | 23 ++ docker/provisioner/app.py | 93 ++++++- 20 files changed, 1081 insertions(+), 115 deletions(-) diff --git a/README.md b/README.md index 9e482c73b..9002ccb4b 100644 --- a/README.md +++ b/README.md @@ -963,7 +963,7 @@ Each task gets its own execution environment with a full filesystem view — ski Uploads from the Web UI, embedded client, and IM channels share one collision-safe storage rule. A completed payload is published only if its candidate name does not exist; concurrent `report.pdf` uploads become `report.pdf`, `report_1.pdf`, `report_2.pdf`, and so on without replacing one another. Collision suffixes remain within the filesystem's 255-byte UTF-8 component limit, including names whose original suffix consumes nearly the whole limit. A busy candidate lease is treated as another collision instead of waiting, so inverse multi-file batches cannot deadlock. The selected name is leased through conversion and sandbox synchronization, so deleting that exact name waits for its active upload lifecycle while unrelated filenames continue concurrently. Internal staging names matching `.upload-*.part` and basenames that cannot be represented losslessly on Windows are rejected for new uploads and reported through `skipped_files`; exact legacy POSIX names already returned by the list endpoint—including literal backslashes and names made only from dots/spaces—remain deletable after upgrade. -Optional document conversions are system-owned assets under `/mnt/user-data/.upload-conversions/`. Normal targets use `.md`; names that would exceed the filesystem component limit use a deterministic UTF-8-safe prefix plus the full SHA-256 digest. The exact generated path is returned through the upload response and omitted from the primary upload listing. Mounted AIO sandboxes expose this namespace through a read-only mount; the remote Provisioner verifies its exact user/thread source and the concrete Pod mount signature. A mount-contract version in deterministic sandbox IDs prevents a pre-upgrade container from satisfying a new acquisition for the same thread; older containers may still be enumerated and adopted only for normal orphan cleanup. Local structured file APIs enforce the same rule through path mappings. Local host bash is outside that mapping boundary and must remain disabled for untrusted tasks. Non-mounted remote providers receive a private synchronized copy that may be writable but cannot mutate the authoritative host conversion or lock state. Direct Markdown uploads use one verified exclusive file descriptor for outline and preview reads, while other formats use only their exact generated asset. Deleting a primary removes its exact host and explicitly synchronized sandbox copies plus its generated asset, and never infers that a user-uploaded sibling such as `uploads/report.md` is disposable. +Optional document conversions are system-owned assets under `/mnt/user-data/.upload-conversions/`. Normal targets use `.md`; names that would exceed the filesystem component limit use a deterministic UTF-8-safe prefix plus the full SHA-256 digest. The exact generated path is returned through the upload response and omitted from the primary upload listing. Mounted AIO sandboxes expose this namespace through a read-only mount; the remote Provisioner verifies its exact user/thread source and the concrete Pod mount signature. A mount-contract version in deterministic sandbox IDs prevents a pre-upgrade container from satisfying a new acquisition for the same thread; older containers may still be enumerated and adopted only for normal orphan cleanup. Local structured file APIs enforce the same rule through path mappings. Local host bash is outside that mapping boundary and must remain disabled for untrusted tasks. Non-mounted remote providers receive a private synchronized copy that may be writable but cannot mutate the authoritative host conversion or lock state. Direct Markdown uploads use one verified exclusive file descriptor for outline and preview reads, while other formats use only their exact generated asset. Deleting a primary removes its exact host and explicitly synchronized sandbox copies plus its generated asset, and never infers that a user-uploaded sibling such as `uploads/report.md` is disposable. Durable deletion of explicit sandbox copies requires a provider to identify both its backend namespace and the immutable sandbox incarnation; AIO Docker/Provisioner backends supply those identities, while providers without this restart-safe capability fail closed before host staging or remote deletion. The built-in `grep` tool searches either one text file or all matching text files below a directory, so an agent can search an uploaded document directly without first broadening the request to the entire uploads directory. @@ -995,6 +995,11 @@ legacy Provisioners cannot isolate equal thread IDs across users. Upgrade the Provisioner first for authenticated deployments. If the capability probe is temporarily unreachable, explicitly requested mounted mode also fails closed instead of creating a Pod whose read-only conversion mount cannot be verified. +Explicit-sync upload deletion additionally requires the Provisioner's immutable +Namespace/Pod identity endpoint. During a mixed-version rollout, upgrade the +Provisioner before relying on deletion of explicitly synchronized sandbox +copies; an old peer stays fail closed rather than treating a missing Service as +proof that its Pod is gone. This is the difference between a chatbot with tool access and an agent with an actual execution environment. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 7c7f0d547..d95a3690f 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -1464,7 +1464,7 @@ Multi-file upload with automatic document conversion: - Exact-name generation leases use a portable NFC-plus-casefold coordination key with Win32 trailing-dot/space folding, so filesystem aliases cannot bypass an active generation; filenames that Windows cannot represent losslessly are rejected before staging, while exact legacy POSIX basenames remain deletable after upgrade. Legacy deletion has a POSIX-only lease mode for literal backslashes and components made entirely from dots/spaces; it does not weaken new-upload validation. The original filename remains the published name. Publication tries each candidate lease without blocking and treats a busy canonical key as a collision, so same-batch and inverse concurrent batches advance to a UTF-8-bounded `_N` candidate instead of deadlocking while retaining earlier generations; pathological long suffixes fall back to truncating the complete basename. Deletion rejects an inode that moved outside the requested name's lease. Primary deletion and rollback move the selected primary and its exact generated conversion into one inode-and-intent-tagged transaction directory under the protected conversion namespace before side effects; the primary lives in a dedicated child directory so legal filenames such as `.commit`, `.restore`, and `.conversion` cannot collide with transaction controls. User deletion starts with restore-on-crash intent. Immediately before the first possible remote mutation, a persisted commit marker makes crash recovery discard the authoritative host tombstones; a live, fully compensated remote failure clears it while restoring both host entries. A live rollback writes a restore marker before publishing either hard link, so recovery can finish restoring both entries even when the transaction began with discard intent. A partial remote failure is compensated from the staged authoritative bytes before host restoration; if compensation fails, every remote target is driven back toward the idempotent deletion outcome and host deletion commits instead of publishing a split generation. Failed-upload rollback tombstones otherwise use discard recovery, while legacy primary-only transactions retain their recorded restore/discard behavior. Neither transaction kind is treated as an ordinary disposable upload stage. Final lease release is the commit point: cancellation newly arriving during release is delayed and swallowed so a committed upload is returned as success rather than an indeterminate cancelled result. Embedded-client finalization attempts every lease release and conversion-pool shutdown independently, logging cleanup failures rather than changing an already-committed response or stranding later leases. - Filenames containing NUL, `<`, `>`, or reserved model-context boundary markers are rejected before staging so accepted filenames and exact virtual paths remain lossless in model-visible upload context. Legacy files discovered on disk are still neutralized when listed. - Gateway HTTP uploads use same-directory `.upload-*.part` staging files. Each active stage holds a cross-process liveness lock under `.upload-conversions/.locks/stages/`; startup cleanup skips held stages and sweeps only crash-orphaned files, while `.upload-delete---.part/` transaction directories are resolved according to their persisted crash intent. Legacy intent-less deletion transactions retain restore behavior. Cancellation during staging creation drains the worker and aborts the returned stage before propagating. Staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools. -- Generated Markdown is owned by `user-data/.upload-conversions/.md` and is omitted from primary upload listings. Deletion holds the generation lease while it stages both authoritative host paths, writes a durable remote-deletion journal, persists the remote-mutation phase, removes an explicitly synchronized sandbox primary/conversion, and commits the host transaction. A remote failure is reported and restores the host generation only after every possibly removed remote path has been re-published; journal-abort errors cannot skip that host restore. Failed compensation instead retries idempotent removal of every remote target and commits host deletion. Unconfirmed paths keep the journal, commit marker, and exact hidden host tombstones; that retained primary persistently reserves its portable filename coordination key, so a new upload advances to `_N` rather than letting an old path-based retry delete the new remote generation. The journal binds both the provider/backend reconciliation namespace and raw sandbox ID. Gateway startup and later embedded deletion preparation may use only the provider's non-creating exact reconnect/confirmed-absence result—never `acquire()`, a replacement sandbox for the thread, or a different provider/backend namespace—before ordinary transaction cleanup removes the tombstones; an unknown lookup stays pending. Live journals remain protected by their stage lease, reconciliation takes the same name lease used by publication, and a disappearing unrelated transaction is treated as completed after the scanner proves it does not reserve the candidate. Exact legacy POSIX basenames use the existing-file deletion validator during retry. It never guesses or deletes a legacy/user-owned `uploads/.md` sibling. Outline extraction opens one descriptor, verifies its `fstat` against the current exclusive regular directory entry, and uses that same descriptor for both outline and preview reads. +- Generated Markdown is owned by `user-data/.upload-conversions/.md` and is omitted from primary upload listings. Deletion holds the generation lease while it stages both authoritative host paths, writes a durable remote-deletion journal, persists the remote-mutation phase, removes an explicitly synchronized sandbox primary/conversion, and commits the host transaction. A remote failure is reported and restores the host generation only after every possibly removed remote path has been re-published; journal-abort errors cannot skip that host restore. Failed compensation instead retries idempotent removal of every remote target and commits host deletion. Unconfirmed paths keep the journal, commit marker, and exact hidden host tombstones; that retained primary persistently reserves its portable filename coordination key, so a new upload advances to `_N` rather than letting an old path-based retry delete the new remote generation. The v3 journal binds the provider type, backend-issued namespace, immutable sandbox incarnation, and raw sandbox ID. A failed directory fsync after unlink recreates the visible journal before returning, preserving that reservation. Gateway startup and later embedded deletion preparation may use only the provider's non-creating exact reconnect/confirmed-absence result—never `acquire()`, a replacement sandbox for the thread, or a different provider/backend/incarnation—before ordinary transaction cleanup removes the tombstones; an unknown lookup stays pending. AIO local Docker uses daemon/container IDs, while the remote Provisioner uses Kubernetes Namespace/Pod UIDs and treats Pod presence as authoritative even when its Service is gone. Providers without an explicit restart-safe reconciliation identity fail deletion preparation before host staging or remote mutation. Live journals remain protected by their stage lease, reconciliation takes the same name lease used by publication, and a disappearing unrelated transaction is treated as completed after the scanner proves it does not reserve the candidate. Exact legacy POSIX basenames use the existing-file deletion validator during retry. It never guesses or deletes a legacy/user-owned `uploads/.md` sibling. Outline extraction opens one descriptor, verifies its `fstat` against the current exclusive regular directory entry, and uses that same descriptor for both outline and preview reads. - Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; only operations that may block waiting for a name lease use the separate lease-wait pool. Work needed by an existing lease holder and non-blocking publication stays on the general pool, so waiters cannot starve conversion, rollback, or release. Cold sandbox-provider construction is also offloaded. Gateway, embedded-client, and IM ingresses share provider-aware publication: mounted providers make the exact host paths sandbox-readable; non-mounted providers acquire the sandbox and synchronize the primary plus generated conversion to their exact virtual paths. Each ingress records attempted remote paths before the write can commit and, on later failure or cancellation, calls the provider-neutral `Sandbox.remove_file()` for those paths before host rollback and lease release; the command fallback requires a per-call unpredictable exact success trailer. Embedded multi-file calls retain every publication and receipt until the whole response is built, then roll back the complete batch on failure. WeChat download publication uses the cancellation-safe async lease adapter, so cancellation drains and rolls back a publication worker that completes late. - Mounted upload paths skip both sandbox acquisition and per-file synchronization. For AIO remote/provisioner deployments this requires an explicit, accurate `sandbox.thread_data_mounts: true` plus a Provisioner advertising a compatible mount contract; every upload/delete/artifact decision waits for any in-flight stale capability probe in a worker thread and uses one atomic snapshot. The same frozen version selects the deterministic sandbox ID, mount set, cache/reuse candidate, create precondition, and response validation; if the peer changes versions during acquisition, the whole attempt restarts with a new ID. The required version is sent in the create JSON, and the Provisioner rejects a mismatch before any Kubernetes read or write, preventing a rolling upgrade from creating an orphan under the prior version's ID. After acquisition, consumers use the immutable contract recorded on that sandbox instance rather than the provider's mutable global capability. Unavailable or legacy results use explicit synchronization and are retried without a Gateway restart. A confirmed legacy peer may create thread sandboxes only for the `default` no-auth user, because its primary user-data mount cannot isolate equal thread IDs across authenticated users; authenticated rolling deployments upgrade the Provisioner first. The Provisioner validates the request before its idempotent fast path, labels Pods with hashed user/thread identity and contract version, stores the exact identity plus a mount-signature annotation, and compares the live Pod specification before reuse. Every remote acquisition path, including active-cache and warm-pool reuse, deliberately replays the idempotent create request instead of trusting GET discovery or a liveness response. Contracts newer than v2 are accepted as forward-compatible, but the response must echo the exact frozen version, user, and thread; legacy responses may omit the version or return exact integer zero, never strings or booleans. The negotiated mount-contract version namespaces deterministic AIO sandbox IDs so a container created under another version cannot satisfy the acquisition; active reuse also replaces its client when validation returns a changed service URL. Reconciliation may still enumerate/adopt the old ID for orphan cleanup; it is not selected for the new identity. - Agent receives uploaded file list via `UploadsMiddleware` diff --git a/backend/docs/FILE_UPLOAD.md b/backend/docs/FILE_UPLOAD.md index 2bece654e..5a70e6d2c 100644 --- a/backend/docs/FILE_UPLOAD.md +++ b/backend/docs/FILE_UPLOAD.md @@ -110,7 +110,9 @@ DELETE /api/threads/{thread_id}/uploads/{filename} } ``` -删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束。对于非挂载 provider，系统在同一个 generation lease 内先暂存宿主机主文件与精确转换副本、持久化远端删除日志，再删除沙箱中的对应副本。远端失败且所有可能的副作用都成功补偿时，请求会返回错误并恢复宿主机文件；即使持久日志的中止清理失败，本地恢复也会在 `finally` 路径完成。如果补偿无法完整确认，系统会重试将全部远端目标收敛到删除状态并提交宿主机删除，避免重新暴露一个远端已经缺失的宿主机 generation，此时请求仍返回错误。持续未确认时，受保护的日志、提交标记和精确宿主机 tombstone 会一起保留；隐藏的主文件 tombstone 持续占用该文件名的可移植协调键，因此新的同名上传会得到 `_N` 名称，不会被旧日志的按路径重试误删。日志同时绑定 provider/backend 命名空间和原始 sandbox ID。Gateway 重启或后续嵌入式删除准备只能通过 provider 的非创建型接口精确重连该旧实例，或由后端明确确认它已经不存在；不会获取线程当前的新 sandbox，也不会跨 provider/backend 命名空间清理。无法确认时继续保留日志与名称预留，远端确认删除或实例确认不存在后才清理 tombstone。随后只删除宿主机上它精确拥有的生成资产和主文件。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞；扫描期间已经完成收尾的无关删除事务会被视为已完成，而不是让当前上传失败。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除，远端日志重试也沿用同一精确旧名称；新上传仍执行严格的跨平台文件名校验。 +删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束。对于非挂载 provider，系统在同一个 generation lease 内先暂存宿主机主文件与精确转换副本、持久化远端删除日志，再删除沙箱中的对应副本。远端失败且所有可能的副作用都成功补偿时，请求会返回错误并恢复宿主机文件；即使持久日志的中止清理失败，本地恢复也会在 `finally` 路径完成。如果补偿无法完整确认，系统会重试将全部远端目标收敛到删除状态并提交宿主机删除，避免重新暴露一个远端已经缺失的宿主机 generation，此时请求仍返回错误。持续未确认时，受保护的日志、提交标记和精确宿主机 tombstone 会一起保留；隐藏的主文件 tombstone 持续占用该文件名的可移植协调键，因此新的同名上传会得到 `_N` 名称，不会被旧日志的按路径重试误删。日志同时绑定 provider 类型、后端不可变命名空间、sandbox 不可变实例 ID 和原始 sandbox ID；日志删除后的目录同步失败会先重建可见日志再返回错误，进程内不会提前释放旧 basename。Gateway 重启或后续嵌入式删除准备只能通过 provider 的非创建型接口精确重连该旧实例，或由同一后端明确确认该实例已经不存在；不会获取线程当前的新 sandbox，也不会跨 provider、后端或同 ID 的新实例清理。AIO 的本地 Docker 使用 daemon/container ID，远端 Provisioner 使用 Kubernetes Namespace/Pod UID；Provisioner 以 Pod 为权威，即使 Service 已经缺失也不会把仍存活的 Pod 报成 absent。其他 provider 若不能提供同等的 restart-safe 身份能力，删除准备会在宿主机暂存和远端删除之前 fail closed。无法确认时继续保留日志与名称预留，远端确认删除或旧实例确认不存在后才清理 tombstone。随后只删除宿主机上它精确拥有的生成资产和主文件。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞；扫描期间已经完成收尾的无关删除事务会被视为已完成，而不是让当前上传失败。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除，远端日志重试也沿用同一精确旧名称；新上传仍执行严格的跨平台文件名校验。 + +如果日志 unlink 已经可见但目录 fsync 失败，系统会先重建日志；即使磁盘错误导致重建也失败，本进程仍会保留 fail-closed 的文件名占用并阻止后台事务清理。重启后，文件系统要么恢复旧日志并继续协调，要么确认日志已经消失并安全释放占用。 ## 支持的文档格式 diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index 2fdcd09a1..0678b6fdc 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -43,7 +43,7 @@ from deerflow.integrations.lark_cli import INTEGRATION_ID as LARK_CLI_INTEGRATIO from deerflow.integrations.lark_cli import LARK_CLI_SANDBOX_CONFIG_DIR, LARK_CLI_SANDBOX_DATA_DIR, LARK_CLI_SANDBOX_RUNTIME_DIR, ensure_lark_cli_credential_tree, lark_skills_installed from deerflow.runtime.user_context import get_effective_user_id from deerflow.sandbox.sandbox import Sandbox -from deerflow.sandbox.sandbox_provider import SandboxProvider, SandboxReconciliationResult +from deerflow.sandbox.sandbox_provider import SandboxProvider, SandboxReconciliationIdentity, SandboxReconciliationResult from deerflow.uploads.layout import UPLOAD_CONVERSIONS_DIRNAME, ensure_conversion_dir from .aio_sandbox import AioSandbox @@ -2572,20 +2572,23 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): return sandbox def reconciliation_provider_key(self) -> str: - """Bind durable journals to this provider's concrete backend namespace.""" + """Return the provider type; the backend namespace is stored separately.""" provider_type = type(self) - backend = self._backend - backend_type = type(backend) - namespace_parts = [ - f"{provider_type.__module__}.{provider_type.__qualname__}", - f"{backend_type.__module__}.{backend_type.__qualname__}", - ] - if isinstance(backend, RemoteSandboxBackend): - namespace_parts.append(backend.provisioner_url) - elif isinstance(backend, LocalContainerBackend): - namespace_parts.extend([backend.runtime, backend._container_prefix]) - namespace = "\0".join(namespace_parts).encode("utf-8") - return f"{namespace_parts[0]}:{hashlib.sha256(namespace).hexdigest()}" + return f"{provider_type.__module__}.{provider_type.__qualname__}" + + def prepare_sandbox_reconciliation_identity( + self, + sandbox_id: str, + ) -> SandboxReconciliationIdentity | None: + """Snapshot a backend-issued namespace and immutable incarnation.""" + discovery = self._backend.discover_for_reconciliation(sandbox_id) + if discovery.status != "found" or discovery.info is None or not discovery.backend_namespace or not discovery.incarnation_id: + return None + return SandboxReconciliationIdentity( + provider_key=self.reconciliation_provider_key(), + backend_namespace=discovery.backend_namespace, + incarnation_id=discovery.incarnation_id, + ) def reconnect_sandbox_for_reconciliation( self, @@ -2593,15 +2596,24 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): *, thread_id: str, user_id: str | None, + identity: SandboxReconciliationIdentity, ) -> SandboxReconciliationResult: """Reconnect to exactly *sandbox_id* without acquiring or creating.""" - active = self.get(sandbox_id) - if active is not None: - return SandboxReconciliationResult.found(active) + if identity.provider_key != self.reconciliation_provider_key(): + return SandboxReconciliationResult.unknown() discovery = self._backend.discover_for_reconciliation(sandbox_id) + if not discovery.backend_namespace or discovery.backend_namespace != identity.backend_namespace: + return SandboxReconciliationResult.unknown() if discovery.status == "absent": return SandboxReconciliationResult.absent() - if discovery.status != "found" or discovery.info is None: + if discovery.status != "found" or not discovery.incarnation_id: + return SandboxReconciliationResult.unknown() + if discovery.incarnation_id != identity.incarnation_id: + # The expected immutable instance is gone. Never redirect the old + # journal to a replacement that reused the deterministic raw ID. + return SandboxReconciliationResult.absent() + if discovery.info is None: + # The exact Pod/container still exists but has no access endpoint. return SandboxReconciliationResult.unknown() info = discovery.info if info.sandbox_id != sandbox_id: @@ -2611,6 +2623,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): effective_user_id = self._effective_acquire_user_id(user_id) if info.user_id is not None and info.user_id != effective_user_id: return SandboxReconciliationResult.unknown() + # Do not borrow an active-cache entry by deterministic raw ID. The + # backend snapshot above is the authority for this exact incarnation; + # a cache entry can be replaced independently during lifecycle races. return SandboxReconciliationResult.found( AioSandbox(sandbox_id, info.sandbox_url), close_after=True, diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/backend.py index 6e20c822c..dc2411a04 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/backend.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/backend.py @@ -25,14 +25,27 @@ class SandboxDiscoveryResult: status: Literal["found", "absent", "unknown"] info: SandboxInfo | None = None + backend_namespace: str | None = None + incarnation_id: str | None = None @classmethod - def found(cls, info: SandboxInfo) -> SandboxDiscoveryResult: - return cls(status="found", info=info) + def found( + cls, + info: SandboxInfo | None, + *, + backend_namespace: str, + incarnation_id: str, + ) -> SandboxDiscoveryResult: + return cls( + status="found", + info=info, + backend_namespace=backend_namespace, + incarnation_id=incarnation_id, + ) @classmethod - def absent(cls) -> SandboxDiscoveryResult: - return cls(status="absent") + def absent(cls, *, backend_namespace: str) -> SandboxDiscoveryResult: + return cls(status="absent", backend_namespace=backend_namespace) @classmethod def unknown(cls) -> SandboxDiscoveryResult: @@ -197,16 +210,13 @@ class SandboxBackend(ABC): def discover_for_reconciliation(self, sandbox_id: str) -> SandboxDiscoveryResult: """Discover one exact old instance without creating a replacement. - The base contract cannot distinguish a definite absence from a - transient discovery failure, so only a positive lookup is conclusive. - Backends with authoritative lookup APIs should override this method. + The base contract has neither a backend namespace nor an immutable + instance identity, so no ordinary discovery result is safe for a + durable mutation. Backends with authoritative identity APIs override + this method. """ - try: - info = self.discover(sandbox_id) - except BaseException: - logger.warning("Sandbox discovery failed during reconciliation", exc_info=True) - return SandboxDiscoveryResult.unknown() - return SandboxDiscoveryResult.found(info) if info is not None else SandboxDiscoveryResult.unknown() + del sandbox_id + return SandboxDiscoveryResult.unknown() def list_running(self) -> list[SandboxInfo]: """Enumerate all running sandboxes managed by this backend. diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py index 3158c0b7c..eba519b8d 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py @@ -408,6 +408,9 @@ class LocalContainerBackend(SandboxBackend): def discover_for_reconciliation(self, sandbox_id: str) -> SandboxDiscoveryResult: """Resolve an exact deterministic container, preserving uncertainty.""" + backend_namespace = self._reconciliation_backend_namespace() + if backend_namespace is None: + return SandboxDiscoveryResult.unknown() container_name = f"{self._container_prefix}-{sandbox_id}" try: running = self._is_container_running(container_name) @@ -415,9 +418,51 @@ class LocalContainerBackend(SandboxBackend): logger.warning("Could not verify container %s during reconciliation", container_name, exc_info=True) return SandboxDiscoveryResult.unknown() if not running: - return SandboxDiscoveryResult.absent() + return SandboxDiscoveryResult.absent(backend_namespace=backend_namespace) info = self.discover(sandbox_id) - return SandboxDiscoveryResult.found(info) if info is not None else SandboxDiscoveryResult.unknown() + incarnation_id = self._container_incarnation(container_name) + if info is None or incarnation_id is None: + return SandboxDiscoveryResult.unknown() + info.container_id = incarnation_id + return SandboxDiscoveryResult.found( + info, + backend_namespace=backend_namespace, + incarnation_id=incarnation_id, + ) + + def _reconciliation_backend_namespace(self) -> str | None: + """Return the Docker daemon identity that owns deterministic names.""" + if self._runtime != "docker": + return None + try: + result = subprocess.run( + [self._runtime, "info", "--format", "{{.ID}}"], + capture_output=True, + text=True, + timeout=5, + ) + except (OSError, subprocess.TimeoutExpired): + return None + daemon_id = result.stdout.strip() + if result.returncode != 0 or not daemon_id or daemon_id == "": + return None + return f"docker:{daemon_id}" + + def _container_incarnation(self, container_name: str) -> str | None: + """Return Docker's immutable container ID for one exact name.""" + try: + result = subprocess.run( + [self._runtime, "inspect", "-f", "{{.Id}}", container_name], + capture_output=True, + text=True, + timeout=5, + ) + except (OSError, subprocess.TimeoutExpired): + return None + incarnation_id = result.stdout.strip() + if result.returncode != 0 or not incarnation_id or incarnation_id == "": + return None + return incarnation_id def list_running(self) -> list[SandboxInfo]: """Enumerate all running containers matching the configured prefix. diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py index 688757af0..56fefe5d1 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py @@ -342,33 +342,52 @@ class RemoteSandboxBackend(SandboxBackend): return self._provisioner_discover(sandbox_id) def discover_for_reconciliation(self, sandbox_id: str) -> SandboxDiscoveryResult: - """Look up an exact provisioner object without mount-version filtering.""" + """Ask the Provisioner for a Pod-authoritative instance identity.""" try: resp = requests.get( - f"{self._provisioner_url}/api/sandboxes/{sandbox_id}", + f"{self._provisioner_url}/api/reconciliation/sandboxes/{sandbox_id}", headers=self._auth_headers(), timeout=10, ) if resp.status_code == 404: - return SandboxDiscoveryResult.absent() + # A rolling old Provisioner has no authoritative endpoint. Its + # ordinary sandbox GET conflates a missing Service with a + # missing Pod and therefore cannot prove absence. + return SandboxDiscoveryResult.unknown() resp.raise_for_status() data = resp.json() if not isinstance(data, dict): raise ValueError("Provisioner reconciliation response is not an object") returned_id = data.get("sandbox_id") - if returned_id is not None and returned_id != sandbox_id: + if returned_id != sandbox_id: raise ValueError("Provisioner reconciliation returned a different sandbox ID") + backend_namespace = data.get("backend_namespace") + if not isinstance(backend_namespace, str) or not backend_namespace: + raise ValueError("Provisioner reconciliation response has no backend namespace") + status = data.get("status") + if status == "absent": + return SandboxDiscoveryResult.absent( + backend_namespace=backend_namespace, + ) + if status != "found": + raise ValueError("Provisioner reconciliation returned an invalid status") + incarnation_id = data.get("incarnation_id") + if not isinstance(incarnation_id, str) or not incarnation_id: + raise ValueError("Provisioner reconciliation response has no incarnation ID") sandbox_url = data.get("sandbox_url") - if not isinstance(sandbox_url, str) or not sandbox_url: - raise ValueError("Provisioner reconciliation response has no sandbox URL") - return SandboxDiscoveryResult.found( - SandboxInfo( + info = None + if isinstance(sandbox_url, str) and sandbox_url: + info = SandboxInfo( sandbox_id=sandbox_id, sandbox_url=sandbox_url, user_id=data.get("user_id"), thread_id=data.get("thread_id"), mount_contract_version=data.get("mount_contract_version"), ) + return SandboxDiscoveryResult.found( + info, + backend_namespace=backend_namespace, + incarnation_id=incarnation_id, ) except (requests.RequestException, ValueError, TypeError): logger.warning("Provisioner reconciliation lookup failed for %s", sandbox_id, exc_info=True) diff --git a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py index 8d0588b12..f686f9451 100644 --- a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py +++ b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py @@ -31,6 +31,15 @@ class SandboxReconciliationResult: return cls(status="unknown") +@dataclass(frozen=True, slots=True) +class SandboxReconciliationIdentity: + """Backend and immutable instance identity persisted by durable operations.""" + + provider_key: str + backend_namespace: str + incarnation_id: str + + class SandboxProvider(ABC): """Abstract base class for sandbox providers""" @@ -66,28 +75,40 @@ class SandboxProvider(ABC): pass def reconciliation_provider_key(self) -> str: - """Return a stable, non-secret backend namespace for durable journals.""" + """Return a stable, non-secret provider type key for durable journals.""" provider_type = type(self) return f"{provider_type.__module__}.{provider_type.__qualname__}" + def prepare_sandbox_reconciliation_identity( + self, + sandbox_id: str, + ) -> SandboxReconciliationIdentity | None: + """Return a restart-safe identity, or ``None`` to disable durable mutation. + + A provider must override this only when it can distinguish both its + backing deployment/account and an immutable sandbox incarnation. The + fail-closed default prevents a deterministic raw ID from being rebound + to a different backend or replacement instance after restart. + """ + del sandbox_id + return None + def reconnect_sandbox_for_reconciliation( self, sandbox_id: str, *, thread_id: str, user_id: str | None, + identity: SandboxReconciliationIdentity, ) -> SandboxReconciliationResult: """Resolve exactly *sandbox_id* without creating or redirecting it. Providers that can discover an old instance or prove it terminally - absent should override this method. The conservative default only - trusts an already-active exact ID; a cache miss remains ``unknown``. + absent should override this method. The conservative default always + remains ``unknown``. """ - del thread_id, user_id - sandbox = self.get(sandbox_id) - if sandbox is None: - return SandboxReconciliationResult.unknown() - return SandboxReconciliationResult.found(sandbox) + del sandbox_id, thread_id, user_id, identity + return SandboxReconciliationResult.unknown() @abstractmethod def release(self, sandbox_id: str) -> None: diff --git a/backend/packages/harness/deerflow/uploads/layout.py b/backend/packages/harness/deerflow/uploads/layout.py index 52d20a259..30810822e 100644 --- a/backend/packages/harness/deerflow/uploads/layout.py +++ b/backend/packages/harness/deerflow/uploads/layout.py @@ -63,6 +63,7 @@ def ensure_conversion_dir(uploads_dir: Path) -> Path: try: conversion_dir.mkdir(mode=0o755) except FileExistsError: + # Validation below rejects an existing non-directory or symlink. pass validated = validate_conversion_dir(uploads_dir) if validated is None: diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index a90603050..9278c4d37 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -10,6 +10,7 @@ import os import secrets import shutil import stat +import threading from collections.abc import Callable, Iterator from dataclasses import dataclass from pathlib import Path, PureWindowsPath @@ -32,6 +33,9 @@ from deerflow.utils.thread_id import validate_thread_id logger = logging.getLogger(__name__) +_uncertain_remote_journal_lock = threading.Lock() +_uncertain_remote_journal_reservations: set[tuple[str, str, str]] = set() + UPLOAD_STAGING_PREFIX = ".upload-" UPLOAD_STAGING_SUFFIX = ".part" UPLOAD_DELETION_TRANSACTION_PREFIX = ".upload-delete-" @@ -53,6 +57,37 @@ class RemoteDeletionCompensatedError(RuntimeError): """Remote deletion failed but its side effects were fully compensated.""" +def _retain_uncertain_remote_journal_reservation(journal_path: Path, filename: str) -> None: + """Reserve a remote pathname while journal-unlink durability is unknown.""" + base_dir = journal_path.parent.parent.parent / "uploads" + reservation = ( + os.path.abspath(base_dir), + portable_name_coordination_key(filename), + os.path.abspath(journal_path), + ) + with _uncertain_remote_journal_lock: + _uncertain_remote_journal_reservations.add(reservation) + + +def _release_uncertain_remote_journal_reservation(journal_path: Path) -> None: + journal_key = os.path.abspath(journal_path) + with _uncertain_remote_journal_lock: + matches = {reservation for reservation in _uncertain_remote_journal_reservations if reservation[2] == journal_key} + _uncertain_remote_journal_reservations.difference_update(matches) + + +def _uncertain_remote_journal_reserves_name(base_dir: Path, coordination_key: str) -> bool: + base_key = os.path.abspath(base_dir) + with _uncertain_remote_journal_lock: + return any(reservation_base == base_key and reservation_name == coordination_key for reservation_base, reservation_name, _journal in _uncertain_remote_journal_reservations) + + +def _transaction_has_uncertain_remote_journal(transaction_dir: Path) -> bool: + transaction_key = os.path.abspath(transaction_dir) + with _uncertain_remote_journal_lock: + return any(os.path.dirname(journal_key) == transaction_key for _base, _name, journal_key in _uncertain_remote_journal_reservations) + + @dataclass(slots=True) class StagedUpload: """A complete-or-in-progress upload stored under a hidden temporary name.""" @@ -307,6 +342,8 @@ def _pending_remote_deletion_reserves_name(base_dir: Path, coordination_key: str Callers hold the candidate's name lease while scanning, so reconciliation cannot remove the journal between this check and publication. """ + if _uncertain_remote_journal_reserves_name(base_dir, coordination_key): + return True conversion_dir = base_dir.parent / UPLOAD_CONVERSIONS_DIRNAME try: conversion_dir_stat = os.lstat(conversion_dir) @@ -1039,6 +1076,8 @@ def _finish_deletion_transaction(staged_path: Path) -> None: def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: """Restore a crash-abandoned tombstone whose basename records its target.""" + if _transaction_has_uncertain_remote_journal(transaction_dir): + return False metadata = _deletion_transaction_metadata(transaction_dir.name) if metadata is None: logger.warning("Refusing malformed upload deletion transaction name: %s", transaction_dir) @@ -1162,10 +1201,6 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: ) return False if not primary_entries: - if remote_journal_entries: - # sandbox_sync reconciles this durable remote cleanup before the - # transaction and its commit marker can be removed. - return False if recover_on_crash and not commit_entries and not conversion_entries: if restore_entries: Path(restore_entries[0].path).unlink() diff --git a/backend/packages/harness/deerflow/uploads/sandbox_sync.py b/backend/packages/harness/deerflow/uploads/sandbox_sync.py index af0de0c16..59ec99479 100644 --- a/backend/packages/harness/deerflow/uploads/sandbox_sync.py +++ b/backend/packages/harness/deerflow/uploads/sandbox_sync.py @@ -15,6 +15,7 @@ from typing import Any from deerflow.config.paths import get_paths from deerflow.sandbox.sandbox_provider import ( + SandboxReconciliationIdentity, SandboxReconciliationResult, sandbox_provider_sandbox_uses_thread_data_mounts, sandbox_provider_uses_thread_data_mounts, @@ -30,6 +31,8 @@ from deerflow.uploads.manager import ( RemoteDeletionCompensatedError, _deletion_transaction_metadata, _normalize_existing_filename, + _release_uncertain_remote_journal_reservation, + _retain_uncertain_remote_journal_reservation, _staged_deletion_remote_journal, cleanup_stale_upload_staging_files, make_upload_file_sandbox_readable, @@ -72,12 +75,70 @@ def _remove_remote_paths(sandbox: Any, virtual_paths: tuple[str, ...]) -> None: def _unlink_journal_durably(journal_path: Path) -> None: - journal_path.unlink(missing_ok=True) - directory_descriptor = os.open(journal_path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) try: - os.fsync(directory_descriptor) - finally: - os.close(directory_descriptor) + journal_data = _read_remote_delete_journal(journal_path) + except FileNotFoundError: + # A prior unlink whose directory fsync failed may still be absent from + # the live namespace yet reappear after a crash. Only a successful + # unlink + parent fsync below may release an in-memory reservation. + return + filename = journal_data.get("filename") + if not isinstance(filename, str): + raise ValueError("remote deletion journal has no filename") + journal_payload = json.dumps( + journal_data, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + _retain_uncertain_remote_journal_reservation(journal_path, filename) + try: + journal_path.unlink(missing_ok=True) + directory_descriptor = os.open(journal_path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(directory_descriptor) + finally: + os.close(directory_descriptor) + _release_uncertain_remote_journal_reservation(journal_path) + except BaseException: + # If unlink became visible but its directory fsync failed, allowing the + # basename reservation to disappear in this process would let a new + # generation reuse the remote path before a crash resurrected the old + # journal. Recreate the same journal entry before propagating. Even if + # the recreation's directory fsync also fails, its visible entry keeps + # publication reserved for the remainder of this process; after a crash + # either a journal survives (reserved) or neither journal does (nothing + # can replay against a later generation). + try: + descriptor = os.open(journal_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + pass + except BaseException: + logger.error("Failed to restore a remote deletion journal after durable unlink failed: %s", journal_path, exc_info=True) + else: + restored = False + try: + view = memoryview(journal_payload) + while view: + written = os.write(descriptor, view) + if written <= 0: + raise OSError("Failed to restore remote deletion journal") + view = view[written:] + os.fsync(descriptor) + restored = True + except BaseException: + logger.error("Failed to rewrite a remote deletion journal after durable unlink failed: %s", journal_path, exc_info=True) + finally: + os.close(descriptor) + if restored: + try: + directory_descriptor = os.open(journal_path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(directory_descriptor) + finally: + os.close(directory_descriptor) + except BaseException: + logger.error("Failed to persist a restored remote deletion journal directory entry: %s", journal_path, exc_info=True) + raise def _read_remote_delete_journal(journal_path: Path) -> dict[str, Any]: @@ -115,6 +176,8 @@ class _SandboxDeletionHook: thread_id: str | None = None user_id: str | None = None provider_key: str | None = None + backend_namespace: str | None = None + incarnation_id: str | None = None _journal_path: Path | None = None _prepared_targets: tuple[tuple[str, bytes | None], ...] | None = None @@ -132,8 +195,8 @@ class _SandboxDeletionHook: conversion_path: Path | None, ) -> None: """Persist enough context to retry an interrupted remote deletion.""" - if not self.sandbox_id or not self.thread_id or not self.provider_key: - raise RuntimeError("Durable remote deletion requires sandbox, thread, and provider identities") + if not self.sandbox_id or not self.thread_id or not self.provider_key or not self.backend_namespace or not self.incarnation_id: + raise RuntimeError("Durable remote deletion requires provider, backend, sandbox, and incarnation identities") targets = ( (upload_virtual_path(filename), primary_path.read_bytes()), ( @@ -144,11 +207,13 @@ class _SandboxDeletionHook: journal_path = _staged_deletion_remote_journal(primary_path) payload = json.dumps( { - "version": 2, + "version": 3, "sandbox_id": self.sandbox_id, "thread_id": self.thread_id, "user_id": self.user_id, "provider_key": self.provider_key, + "backend_namespace": self.backend_namespace, + "incarnation_id": self.incarnation_id, "filename": filename, "virtual_paths": [path for path, _bytes in targets], }, @@ -267,6 +332,8 @@ def _deletion_hook_for_sandbox( thread_id: str | None = None, user_id: str | None = None, provider_key: str | None = None, + backend_namespace: str | None = None, + incarnation_id: str | None = None, ) -> _SandboxDeletionHook: return _SandboxDeletionHook( sandbox=sandbox, @@ -274,6 +341,8 @@ def _deletion_hook_for_sandbox( thread_id=thread_id, user_id=user_id, provider_key=provider_key, + backend_namespace=backend_namespace, + incarnation_id=incarnation_id, ) @@ -288,7 +357,7 @@ def _declared_callable(instance: Any, name: str) -> Callable[..., Any] | None: def _sandbox_provider_reconciliation_key(sandbox_provider: Any) -> str: - """Return the durable namespace that owns a provider's raw sandbox IDs.""" + """Return the stable provider type key recorded by durable journals.""" key_factory = _declared_callable(sandbox_provider, "reconciliation_provider_key") if key_factory is None: provider_type = type(sandbox_provider) @@ -300,12 +369,36 @@ def _sandbox_provider_reconciliation_key(sandbox_provider: Any) -> str: return key +def _prepare_sandbox_reconciliation_identity( + sandbox_provider: Any, + sandbox_id: str, +) -> SandboxReconciliationIdentity: + identity_factory = _declared_callable( + sandbox_provider, + "prepare_sandbox_reconciliation_identity", + ) + identity = identity_factory(sandbox_id) if identity_factory is not None else None + if not isinstance(identity, SandboxReconciliationIdentity): + raise RuntimeError("The active sandbox provider cannot establish a restart-safe backend and incarnation identity for upload deletion") + if identity.provider_key != _sandbox_provider_reconciliation_key(sandbox_provider): + raise ValueError("Sandbox reconciliation identity does not match its provider") + for field in ( + identity.provider_key, + identity.backend_namespace, + identity.incarnation_id, + ): + if not isinstance(field, str) or not field or len(field.encode("utf-8")) > 512: + raise ValueError("Sandbox reconciliation identity fields must be non-empty strings of at most 512 bytes") + return identity + + def _reconnect_sandbox_for_reconciliation( sandbox_provider: Any, sandbox_id: str, *, thread_id: str, user_id: str | None, + identity: SandboxReconciliationIdentity, ) -> SandboxReconciliationResult: """Resolve one exact old sandbox without creating or redirecting it.""" resolver = _declared_callable(sandbox_provider, "reconnect_sandbox_for_reconciliation") @@ -314,10 +407,10 @@ def _reconnect_sandbox_for_reconciliation( sandbox_id, thread_id=thread_id, user_id=user_id, + identity=identity, ) else: - sandbox = sandbox_provider.get(sandbox_id) - result = SandboxReconciliationResult.found(sandbox) if sandbox is not None else SandboxReconciliationResult.unknown() + result = SandboxReconciliationResult.unknown() if not isinstance(result, SandboxReconciliationResult): raise TypeError("Sandbox reconciliation returned an invalid result") if result.status == "found" and result.sandbox is None: @@ -387,13 +480,15 @@ def reconcile_pending_remote_deletions( try: try: data = _read_remote_delete_journal(journal_path) - if data.get("version") != 2: + if data.get("version") != 3: raise ValueError("unsupported journal version") filename = data.get("filename") thread_id = data.get("thread_id") user_id = data.get("user_id") sandbox_id = data.get("sandbox_id") provider_key = data.get("provider_key") + backend_namespace = data.get("backend_namespace") + incarnation_id = data.get("incarnation_id") virtual_paths = data.get("virtual_paths") expected_paths = [ upload_virtual_path(filename) if isinstance(filename, str) else None, @@ -409,6 +504,12 @@ def reconcile_pending_remote_deletions( or not isinstance(provider_key, str) or not provider_key or len(provider_key.encode("utf-8")) > 512 + or not isinstance(backend_namespace, str) + or not backend_namespace + or len(backend_namespace.encode("utf-8")) > 512 + or not isinstance(incarnation_id, str) + or not incarnation_id + or len(incarnation_id.encode("utf-8")) > 512 or virtual_paths != expected_paths ): raise ValueError("invalid remote deletion journal fields") @@ -424,6 +525,11 @@ def reconcile_pending_remote_deletions( provider_key, ) continue + identity = SandboxReconciliationIdentity( + provider_key=provider_key, + backend_namespace=backend_namespace, + incarnation_id=incarnation_id, + ) name_lease = UploadNameLease.try_acquire( uploads_dir, filename, @@ -437,6 +543,7 @@ def reconcile_pending_remote_deletions( sandbox_id, thread_id=thread_id, user_id=user_id, + identity=identity, ) if result.status == "unknown": logger.warning( @@ -496,12 +603,18 @@ def prepare_upload_deletion( sandbox = sandbox_provider.get(sandbox_id) if sandbox is None: raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire") + identity = _prepare_sandbox_reconciliation_identity( + sandbox_provider, + sandbox_id, + ) return _deletion_hook_for_sandbox( sandbox, sandbox_id=sandbox_id, thread_id=thread_id, user_id=user_id, - provider_key=_sandbox_provider_reconciliation_key(sandbox_provider), + provider_key=identity.provider_key, + backend_namespace=identity.backend_namespace, + incarnation_id=identity.incarnation_id, ) @@ -526,12 +639,19 @@ async def prepare_upload_deletion_async( sandbox = sandbox_provider.get(sandbox_id) if sandbox is None: raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire") + identity = await asyncio.to_thread( + _prepare_sandbox_reconciliation_identity, + sandbox_provider, + sandbox_id, + ) return _deletion_hook_for_sandbox( sandbox, sandbox_id=sandbox_id, thread_id=thread_id, user_id=user_id, - provider_key=_sandbox_provider_reconciliation_key(sandbox_provider), + provider_key=identity.provider_key, + backend_namespace=identity.backend_namespace, + incarnation_id=identity.incarnation_id, ) diff --git a/backend/tests/test_aio_sandbox_local_backend.py b/backend/tests/test_aio_sandbox_local_backend.py index d5c5f1ab4..07a66c192 100644 --- a/backend/tests/test_aio_sandbox_local_backend.py +++ b/backend/tests/test_aio_sandbox_local_backend.py @@ -312,16 +312,19 @@ def test_discover_returns_none_when_runtime_check_times_out(monkeypatch): def test_reconciliation_discovery_confirms_exact_container_absent(monkeypatch): backend = _backend_for_inspect_tests() + monkeypatch.setattr(backend, "_reconciliation_backend_namespace", lambda: "docker:daemon-1") monkeypatch.setattr(backend, "_is_container_running", lambda _name: False) result = backend.discover_for_reconciliation("missing") assert result.status == "absent" assert result.info is None + assert result.backend_namespace == "docker:daemon-1" def test_reconciliation_discovery_preserves_runtime_uncertainty(monkeypatch): backend = _backend_for_inspect_tests() + monkeypatch.setattr(backend, "_reconciliation_backend_namespace", lambda: "docker:daemon-1") def fail(_name): raise RuntimeError("daemon unavailable") @@ -339,13 +342,46 @@ def test_reconciliation_discovery_returns_exact_running_container(monkeypatch): backend = _backend_for_inspect_tests() info = SandboxInfo(sandbox_id="existing", sandbox_url="http://localhost:18080") + monkeypatch.setattr(backend, "_reconciliation_backend_namespace", lambda: "docker:daemon-1") monkeypatch.setattr(backend, "_is_container_running", lambda _name: True) monkeypatch.setattr(backend, "discover", lambda sandbox_id: info if sandbox_id == "existing" else None) + monkeypatch.setattr(backend, "_container_incarnation", lambda _name: "container-uid-1") result = backend.discover_for_reconciliation("existing") assert result.status == "found" assert result.info is info + assert result.backend_namespace == "docker:daemon-1" + assert result.incarnation_id == "container-uid-1" + + +def test_reconciliation_discovery_fails_closed_without_docker_daemon_identity(monkeypatch): + backend = _backend_for_inspect_tests() + monkeypatch.setattr(backend, "_reconciliation_backend_namespace", lambda: None) + running = SimpleNamespace(called=False) + + def should_not_inspect(_name): + running.called = True + raise AssertionError("container lookup must not cross an unknown daemon boundary") + + monkeypatch.setattr(backend, "_is_container_running", should_not_inspect) + + result = backend.discover_for_reconciliation("same-id") + + assert result.status == "unknown" + assert running.called is False + + +@pytest.mark.parametrize("value", ["", ""]) +def test_reconciliation_backend_namespace_rejects_missing_docker_daemon_id(monkeypatch, value): + backend = _backend_for_inspect_tests() + monkeypatch.setattr( + subprocess, + "run", + lambda *_args, **_kwargs: SimpleNamespace(returncode=0, stdout=f"{value}\n"), + ) + + assert backend._reconciliation_backend_namespace() is None def test_is_container_running_false_on_apple_container_not_found(monkeypatch): diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index a34993fff..808057152 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -2030,6 +2030,7 @@ def test_reconciliation_reconnects_exact_discovered_sandbox_without_acquiring(mo aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") backend_mod = importlib.import_module("deerflow.community.aio_sandbox.backend") info_mod = importlib.import_module("deerflow.community.aio_sandbox.sandbox_info") + provider_mod = importlib.import_module("deerflow.sandbox.sandbox_provider") provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) info = info_mod.SandboxInfo( sandbox_id="old-id", @@ -2038,9 +2039,13 @@ def test_reconciliation_reconnects_exact_discovered_sandbox_without_acquiring(mo user_id="alice", ) provider._backend = SimpleNamespace( - discover_for_reconciliation=lambda sandbox_id: backend_mod.SandboxDiscoveryResult.found(info), + discover_for_reconciliation=lambda sandbox_id: backend_mod.SandboxDiscoveryResult.found( + info, + backend_namespace="namespace-1", + incarnation_id="pod-1", + ), ) - provider.get = lambda _sandbox_id: None + provider.get = lambda _sandbox_id: (_ for _ in ()).throw(AssertionError("reconciliation must not borrow the active cache")) created: list[tuple[str, str]] = [] class TransientSandbox: @@ -2053,6 +2058,11 @@ def test_reconciliation_reconnects_exact_discovered_sandbox_without_acquiring(mo "old-id", thread_id="thread-1", user_id="alice", + identity=provider_mod.SandboxReconciliationIdentity( + provider_key=provider.reconciliation_provider_key(), + backend_namespace="namespace-1", + incarnation_id="pod-1", + ), ) assert result.status == "found" @@ -2064,6 +2074,7 @@ def test_reconciliation_rejects_discovered_sandbox_from_another_identity(): aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") backend_mod = importlib.import_module("deerflow.community.aio_sandbox.backend") info_mod = importlib.import_module("deerflow.community.aio_sandbox.sandbox_info") + provider_mod = importlib.import_module("deerflow.sandbox.sandbox_provider") provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) info = info_mod.SandboxInfo( sandbox_id="old-id", @@ -2072,7 +2083,11 @@ def test_reconciliation_rejects_discovered_sandbox_from_another_identity(): user_id="alice", ) provider._backend = SimpleNamespace( - discover_for_reconciliation=lambda sandbox_id: backend_mod.SandboxDiscoveryResult.found(info), + discover_for_reconciliation=lambda sandbox_id: backend_mod.SandboxDiscoveryResult.found( + info, + backend_namespace="namespace-1", + incarnation_id="pod-1", + ), ) provider.get = lambda _sandbox_id: None @@ -2080,18 +2095,139 @@ def test_reconciliation_rejects_discovered_sandbox_from_another_identity(): "old-id", thread_id="thread-1", user_id="alice", + identity=provider_mod.SandboxReconciliationIdentity( + provider_key=provider.reconciliation_provider_key(), + backend_namespace="namespace-1", + incarnation_id="pod-1", + ), ) assert result.status == "unknown" assert result.sandbox is None -def test_reconciliation_provider_key_changes_with_remote_backend_namespace(): +def test_reconciliation_identity_separates_provider_type_from_backend_namespace(): aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") - remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend") + backend_mod = importlib.import_module("deerflow.community.aio_sandbox.backend") + info_mod = importlib.import_module("deerflow.community.aio_sandbox.sandbox_info") first = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) second = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) - first._backend = remote_mod.RemoteSandboxBackend("http://provisioner-a:8002") - second._backend = remote_mod.RemoteSandboxBackend("http://provisioner-b:8002") + info = info_mod.SandboxInfo(sandbox_id="same-id", sandbox_url="http://sandbox") + first._backend = SimpleNamespace( + discover_for_reconciliation=lambda _sandbox_id: backend_mod.SandboxDiscoveryResult.found( + info, + backend_namespace="cluster-a", + incarnation_id="pod-a", + ) + ) + second._backend = SimpleNamespace( + discover_for_reconciliation=lambda _sandbox_id: backend_mod.SandboxDiscoveryResult.found( + info, + backend_namespace="cluster-b", + incarnation_id="pod-b", + ) + ) - assert first.reconciliation_provider_key() != second.reconciliation_provider_key() + first_identity = first.prepare_sandbox_reconciliation_identity("same-id") + second_identity = second.prepare_sandbox_reconciliation_identity("same-id") + + assert first.reconciliation_provider_key() == second.reconciliation_provider_key() + assert first_identity is not None + assert second_identity is not None + assert first_identity.backend_namespace == "cluster-a" + assert second_identity.backend_namespace == "cluster-b" + + +def test_reconciliation_never_uses_active_replacement_with_same_raw_id(): + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + backend_mod = importlib.import_module("deerflow.community.aio_sandbox.backend") + info_mod = importlib.import_module("deerflow.community.aio_sandbox.sandbox_info") + provider_mod = importlib.import_module("deerflow.sandbox.sandbox_provider") + provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) + replacement = MagicMock() + info = info_mod.SandboxInfo( + sandbox_id="same-id", + sandbox_url="http://replacement", + thread_id="thread-1", + user_id="alice", + ) + provider._backend = SimpleNamespace( + discover_for_reconciliation=lambda _sandbox_id: backend_mod.SandboxDiscoveryResult.found( + info, + backend_namespace="namespace-1", + incarnation_id="pod-new", + ) + ) + provider.get = lambda _sandbox_id: replacement + + result = provider.reconnect_sandbox_for_reconciliation( + "same-id", + thread_id="thread-1", + user_id="alice", + identity=provider_mod.SandboxReconciliationIdentity( + provider_key=provider.reconciliation_provider_key(), + backend_namespace="namespace-1", + incarnation_id="pod-old", + ), + ) + + assert result.status == "absent" + assert result.sandbox is None + replacement.remove_file.assert_not_called() + + +def test_reconciliation_keeps_journal_when_backend_namespace_changes(): + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + backend_mod = importlib.import_module("deerflow.community.aio_sandbox.backend") + provider_mod = importlib.import_module("deerflow.sandbox.sandbox_provider") + provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) + provider._backend = SimpleNamespace(discover_for_reconciliation=lambda _sandbox_id: backend_mod.SandboxDiscoveryResult.absent(backend_namespace="cluster-new")) + + result = provider.reconnect_sandbox_for_reconciliation( + "same-id", + thread_id="thread-1", + user_id="alice", + identity=provider_mod.SandboxReconciliationIdentity( + provider_key=provider.reconciliation_provider_key(), + backend_namespace="cluster-old", + incarnation_id="pod-old", + ), + ) + + assert result.status == "unknown" + + +def test_reconciliation_keeps_journal_when_exact_pod_has_no_service(): + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + backend_mod = importlib.import_module("deerflow.community.aio_sandbox.backend") + provider_mod = importlib.import_module("deerflow.sandbox.sandbox_provider") + provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) + provider._backend = SimpleNamespace( + discover_for_reconciliation=lambda _sandbox_id: backend_mod.SandboxDiscoveryResult.found( + None, + backend_namespace="namespace-1", + incarnation_id="pod-1", + ) + ) + + result = provider.reconnect_sandbox_for_reconciliation( + "same-id", + thread_id="thread-1", + user_id="alice", + identity=provider_mod.SandboxReconciliationIdentity( + provider_key=provider.reconciliation_provider_key(), + backend_namespace="namespace-1", + incarnation_id="pod-1", + ), + ) + + assert result.status == "unknown" + + +def test_prepare_reconciliation_identity_fails_closed_without_exact_instance(): + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + backend_mod = importlib.import_module("deerflow.community.aio_sandbox.backend") + provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) + provider._backend = SimpleNamespace(discover_for_reconciliation=lambda _sandbox_id: backend_mod.SandboxDiscoveryResult.unknown()) + + assert provider.prepare_sandbox_reconciliation_identity("same-id") is None diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index 946eb2773..a23844968 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -28,6 +28,7 @@ from deerflow.client import DeerFlowClient from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig from deerflow.config.paths import Paths +from deerflow.sandbox.sandbox_provider import SandboxReconciliationIdentity from deerflow.skills.types import SkillCategory from deerflow.tools.mcp_metadata import tag_mcp_tool from deerflow.uploads.layout import conversion_path_for_upload @@ -2671,10 +2672,36 @@ class TestUploads: conversion = conversion_path_for_upload(primary) conversion.parent.mkdir() conversion.write_text("generated", encoding="utf-8") - provider = MagicMock(uses_thread_data_mounts=False) - provider.acquire.return_value = "remote-1" sandbox = MagicMock() - provider.get.return_value = sandbox + + class Provider: + uses_thread_data_mounts = False + + @staticmethod + def refresh_thread_data_mount_capabilities() -> bool: + return False + + @staticmethod + def acquire(_thread_id: str, *, user_id: str | None = None) -> str: + return "remote-1" + + @staticmethod + def get(sandbox_id: str): + return sandbox if sandbox_id == "remote-1" else None + + @staticmethod + def reconciliation_provider_key() -> str: + return "tests.ClientExplicitSyncProvider" + + def prepare_sandbox_reconciliation_identity(self, sandbox_id: str) -> SandboxReconciliationIdentity: + assert sandbox_id == "remote-1" + return SandboxReconciliationIdentity( + provider_key=self.reconciliation_provider_key(), + backend_namespace="tests.backend", + incarnation_id="tests.incarnation", + ) + + provider = Provider() with ( patch("deerflow.client.get_uploads_dir", return_value=uploads_dir), diff --git a/backend/tests/test_provisioner_request_threading.py b/backend/tests/test_provisioner_request_threading.py index 3629c0264..933b2ad61 100644 --- a/backend/tests/test_provisioner_request_threading.py +++ b/backend/tests/test_provisioner_request_threading.py @@ -86,7 +86,10 @@ class _RecordingCoreV1: self.created_pods: list[str] = [] self.created_pod_specs: dict[str, object] = {} self.existing_pod_specs: dict[str, object] = {} + self.missing_pods: set[str] = set() + self.pod_read_counts: dict[str, int] = {} self.created_services: list[str] = [] + self.namespace_uid = "namespace-uid-1" def _record_k8s_call(self) -> None: thread_id = threading.get_ident() @@ -115,12 +118,21 @@ class _RecordingCoreV1: def read_namespaced_pod(self, _name: str, _namespace: str): self._record_k8s_call() sandbox_id = _name.removeprefix("sandbox-") + self.pod_read_counts[sandbox_id] = self.pod_read_counts.get(sandbox_id, 0) + 1 + if sandbox_id in self.missing_pods: + raise ApiException(status=404) pod = self.existing_pod_specs.get(sandbox_id) or self.created_pod_specs.get(sandbox_id) if pod is not None: + if not getattr(pod.metadata, "uid", None): + pod.metadata.uid = f"pod-uid-{sandbox_id}" pod.status = SimpleNamespace(phase="Running") return pod return SimpleNamespace(status=SimpleNamespace(phase="Running")) + def read_namespace(self, _namespace: str): + self._record_k8s_call() + return SimpleNamespace(metadata=SimpleNamespace(uid=self.namespace_uid)) + def create_namespaced_pod(self, _namespace: str, pod) -> None: self._record_k8s_call() sandbox_id = pod.metadata.labels["sandbox-id"] @@ -173,6 +185,7 @@ def test_sandbox_business_route_handlers_are_sync(provisioner_module) -> None: for handler in ( provisioner_module.create_sandbox, provisioner_module.destroy_sandbox, + provisioner_module.reconcile_sandbox_identity, provisioner_module.get_sandbox, provisioner_module.list_sandboxes, ): @@ -227,6 +240,104 @@ async def test_sandbox_business_routes_run_k8s_client_off_event_loop_thread( assert fake_core_v1.created_services == [expected_created_sandbox] +@pytest.mark.asyncio +async def test_reconciliation_endpoint_reports_pod_absence_authoritatively( + monkeypatch: pytest.MonkeyPatch, + provisioner_module, +) -> None: + fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=threading.get_ident()) + fake_core_v1.missing_pods.add("sandbox-gone") + monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1) + monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret") + + with _detect_provisioner_blocking_io(provisioner_module): + transport = httpx.ASGITransport(app=provisioner_module.app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.get( + "/api/reconciliation/sandboxes/sandbox-gone", + headers={"X-API-Key": "test-secret"}, + ) + + assert response.status_code == 200 + assert response.json() == { + "sandbox_id": "sandbox-gone", + "status": "absent", + "backend_namespace": "namespace-uid-1", + "incarnation_id": None, + "sandbox_url": None, + "user_id": None, + "thread_id": None, + "mount_contract_version": None, + } + + +@pytest.mark.asyncio +async def test_reconciliation_endpoint_keeps_live_pod_when_service_is_missing( + monkeypatch: pytest.MonkeyPatch, + provisioner_module, +) -> None: + fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=threading.get_ident()) + fake_core_v1.existing_pod_specs["sandbox-partial"] = provisioner_module._build_pod( + "sandbox-partial", + "thread-1", + user_id="alice", + ) + monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1) + monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret") + + with _detect_provisioner_blocking_io(provisioner_module): + transport = httpx.ASGITransport(app=provisioner_module.app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.get( + "/api/reconciliation/sandboxes/sandbox-partial", + headers={"X-API-Key": "test-secret"}, + ) + + assert response.status_code == 200 + payload = response.json() + assert payload["status"] == "found" + assert payload["backend_namespace"] == "namespace-uid-1" + assert payload["incarnation_id"] == "pod-uid-sandbox-partial" + assert payload["sandbox_url"] is None + assert payload["user_id"] == "alice" + assert payload["thread_id"] == "thread-1" + assert fake_core_v1.pod_read_counts["sandbox-partial"] == 1 + + +@pytest.mark.asyncio +async def test_reconciliation_endpoint_returns_service_url_for_live_pod( + monkeypatch: pytest.MonkeyPatch, + provisioner_module, +) -> None: + fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=threading.get_ident()) + fake_core_v1.existing_pod_specs["sandbox-live"] = provisioner_module._build_pod( + "sandbox-live", + "thread-1", + user_id="alice", + ) + fake_core_v1.service_sandboxes.add("sandbox-live") + monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1) + monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret") + + with _detect_provisioner_blocking_io(provisioner_module): + transport = httpx.ASGITransport(app=provisioner_module.app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.get( + "/api/reconciliation/sandboxes/sandbox-live", + headers={"X-API-Key": "test-secret"}, + ) + + assert response.status_code == 200 + payload = response.json() + assert payload["status"] == "found" + assert payload["incarnation_id"] == "pod-uid-sandbox-live" + assert payload["sandbox_url"] == provisioner_module._sandbox_url( + "sandbox-live", + node_port=32123, + ) + assert fake_core_v1.pod_read_counts["sandbox-live"] == 1 + + def test_existing_sandbox_rejects_cross_tenant_reuse( monkeypatch: pytest.MonkeyPatch, provisioner_module, diff --git a/backend/tests/test_remote_sandbox_backend.py b/backend/tests/test_remote_sandbox_backend.py index fcac847af..b9e2f41ab 100644 --- a/backend/tests/test_remote_sandbox_backend.py +++ b/backend/tests/test_remote_sandbox_backend.py @@ -51,16 +51,36 @@ def test_list_running_delegates_to_provisioner_list(monkeypatch): assert backend.list_running() == [sandbox_info] -def test_reconciliation_discovery_confirms_provisioner_absence(monkeypatch): +def test_reconciliation_discovery_does_not_trust_legacy_route_404(monkeypatch): backend = RemoteSandboxBackend("http://provisioner:8002") monkeypatch.setattr(requests, "get", lambda *_args, **_kwargs: _StubResponse(status_code=404)) result = backend.discover_for_reconciliation("old-id") - assert result.status == "absent" + assert result.status == "unknown" assert result.info is None +def test_reconciliation_discovery_accepts_pod_authoritative_absence(monkeypatch): + backend = RemoteSandboxBackend("http://provisioner:8002") + monkeypatch.setattr( + requests, + "get", + lambda *_args, **_kwargs: _StubResponse( + payload={ + "sandbox_id": "old-id", + "status": "absent", + "backend_namespace": "namespace-uid-1", + } + ), + ) + + result = backend.discover_for_reconciliation("old-id") + + assert result.status == "absent" + assert result.backend_namespace == "namespace-uid-1" + + def test_reconciliation_discovery_preserves_transport_uncertainty(monkeypatch): backend = RemoteSandboxBackend("http://provisioner:8002") @@ -75,7 +95,7 @@ def test_reconciliation_discovery_preserves_transport_uncertainty(monkeypatch): assert result.info is None -def test_reconciliation_discovery_accepts_exact_legacy_mount_contract(monkeypatch): +def test_reconciliation_discovery_accepts_exact_old_mount_contract(monkeypatch): backend = RemoteSandboxBackend("http://provisioner:8002") monkeypatch.setattr( requests, @@ -83,6 +103,9 @@ def test_reconciliation_discovery_accepts_exact_legacy_mount_contract(monkeypatc lambda *_args, **_kwargs: _StubResponse( payload={ "sandbox_id": "old-id", + "status": "found", + "backend_namespace": "namespace-uid-1", + "incarnation_id": "pod-uid-1", "sandbox_url": "http://sandbox-old", "thread_id": "thread-1", "user_id": "alice", @@ -97,6 +120,8 @@ def test_reconciliation_discovery_accepts_exact_legacy_mount_contract(monkeypatc assert result.info is not None assert result.info.sandbox_id == "old-id" assert result.info.mount_contract_version == 1 + assert result.backend_namespace == "namespace-uid-1" + assert result.incarnation_id == "pod-uid-1" def test_provisioner_list_returns_sandbox_infos_and_filters_invalid_entries(monkeypatch): diff --git a/backend/tests/test_upload_sandbox_sync.py b/backend/tests/test_upload_sandbox_sync.py index 02dd38206..7c868cf2a 100644 --- a/backend/tests/test_upload_sandbox_sync.py +++ b/backend/tests/test_upload_sandbox_sync.py @@ -6,6 +6,7 @@ from unittest.mock import patch import pytest +from deerflow.sandbox.sandbox_provider import SandboxReconciliationResult from deerflow.uploads.layout import conversion_path_for_upload, conversion_virtual_path, upload_virtual_path from deerflow.uploads.manager import ( RemoteDeletionCommitRequiredError, @@ -14,7 +15,9 @@ from deerflow.uploads.manager import ( publish_upload_bytes, ) from deerflow.uploads.sandbox_sync import ( - _deletion_hook_for_sandbox, + _deletion_hook_for_sandbox as _raw_deletion_hook_for_sandbox, +) +from deerflow.uploads.sandbox_sync import ( _sandbox_provider_reconciliation_key, make_upload_paths_available, make_upload_paths_available_async, @@ -24,6 +27,29 @@ from deerflow.uploads.sandbox_sync import ( ) +def _deletion_hook_for_sandbox(sandbox, **kwargs): + if kwargs.get("sandbox_id") is not None: + kwargs.setdefault("backend_namespace", "tests.backend") + kwargs.setdefault("incarnation_id", "tests.incarnation") + return _raw_deletion_hook_for_sandbox(sandbox, **kwargs) + + +class _ExactProviderMixin: + def reconnect_sandbox_for_reconciliation( + self, + sandbox_id, + *, + thread_id, + user_id, + identity, + ): + del thread_id, user_id + assert identity.backend_namespace == "tests.backend" + assert identity.incarnation_id == "tests.incarnation" + sandbox = self.get(sandbox_id) + return SandboxReconciliationResult.found(sandbox) if sandbox is not None else SandboxReconciliationResult.unknown() + + def test_failed_compensation_deletes_paths_that_were_already_republished(tmp_path): primary_virtual = "/mnt/user-data/uploads/report.pdf" conversion_virtual = "/mnt/user-data/.upload-conversions/report.pdf.md" @@ -91,7 +117,7 @@ def test_unconfirmed_remote_delete_is_persisted_and_retried_after_restart(tmp_pa def update_file(self, _path: str, _content: bytes) -> None: raise OSError("compensation unavailable") - class Provider: + class Provider(_ExactProviderMixin): def __init__(self, sandbox) -> None: self.sandbox = sandbox @@ -194,6 +220,163 @@ def test_remote_delete_finalize_failure_never_restores_deleted_host_generation(t assert cleanup_stale_upload_staging_files(tmp_path) == 0 +def test_directory_fsync_failure_restores_journal_and_reserves_old_basename(tmp_path): + import deerflow.uploads.manager as upload_manager_module + import deerflow.uploads.sandbox_sync as sandbox_sync_module + + primary_virtual = upload_virtual_path("report.pdf") + conversion_virtual = conversion_virtual_path("report.pdf") + + class Sandbox: + def __init__(self) -> None: + self.files = { + primary_virtual: b"old primary", + conversion_virtual: b"old conversion", + } + + def remove_file(self, path: str) -> None: + self.files.pop(path, None) + + class Provider(_ExactProviderMixin): + def __init__(self, sandbox) -> None: + self.sandbox = sandbox + + def get(self, sandbox_id: str): + assert sandbox_id == "sandbox-1" + return self.sandbox + + uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads" + uploads.mkdir(parents=True) + primary = uploads / "report.pdf" + primary.write_bytes(b"old primary") + conversion = conversion_path_for_upload(primary) + conversion.parent.mkdir() + conversion.write_bytes(b"old conversion") + identity = upload_manager_module.UploadIdentity.from_path(primary) + staged_path, stage_lease = upload_manager_module._stage_primary_deletion( + uploads, + primary, + identity, + recover_on_crash=True, + conversion_path=conversion, + ) + sandbox = Sandbox() + provider = Provider(sandbox) + hook = _deletion_hook_for_sandbox( + sandbox, + sandbox_id="sandbox-1", + thread_id="thread-1", + user_id="alice", + provider_key=_sandbox_provider_reconciliation_key(provider), + ) + hook.prepare( + primary.name, + staged_path, + upload_manager_module._staged_conversion_path(staged_path), + ) + upload_manager_module._mark_staged_deletion_committed(staged_path) + stage_lease.release() + journal_path = next(conversion.parent.glob(".upload-delete-*.part/.remote-delete.json")) + journal_payload = journal_path.read_bytes() + real_fsync = sandbox_sync_module.os.fsync + fsync_calls = 0 + + def fail_first_directory_fsync(descriptor): + nonlocal fsync_calls + fsync_calls += 1 + if fsync_calls == 1: + raise OSError("directory fsync failed") + return real_fsync(descriptor) + + with patch.object(sandbox_sync_module.os, "fsync", side_effect=fail_first_directory_fsync): + with pytest.raises(OSError, match="directory fsync failed"): + sandbox_sync_module._unlink_journal_durably(journal_path) + + assert journal_path.read_bytes() == journal_payload + replacement = publish_upload_bytes(uploads, "report.pdf", b"new primary") + assert replacement.name == "report_1.pdf" + + assert ( + reconcile_pending_remote_deletions( + sandbox_provider_factory=lambda: provider, + base_dir=tmp_path, + ) + == 1 + ) + assert sandbox.files == {} + assert replacement.read_bytes() == b"new primary" + assert cleanup_stale_upload_staging_files(tmp_path) == 1 + + +def test_failed_journal_restore_keeps_in_memory_name_reservation(tmp_path): + import deerflow.uploads.manager as upload_manager_module + import deerflow.uploads.sandbox_sync as sandbox_sync_module + + uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads" + uploads.mkdir(parents=True) + primary = uploads / "report.pdf" + primary.write_bytes(b"old primary") + conversion = conversion_path_for_upload(primary) + conversion.parent.mkdir() + conversion.write_bytes(b"old conversion") + identity = upload_manager_module.UploadIdentity.from_path(primary) + staged_path, stage_lease = upload_manager_module._stage_primary_deletion( + uploads, + primary, + identity, + recover_on_crash=True, + conversion_path=conversion, + ) + hook = _deletion_hook_for_sandbox( + object(), + sandbox_id="sandbox-1", + thread_id="thread-1", + user_id="alice", + provider_key="tests.Provider", + ) + hook.prepare( + primary.name, + staged_path, + upload_manager_module._staged_conversion_path(staged_path), + ) + upload_manager_module._mark_staged_deletion_committed(staged_path) + stage_lease.release() + journal_path = next(conversion.parent.glob(".upload-delete-*.part/.remote-delete.json")) + real_open = sandbox_sync_module.os.open + real_fsync = sandbox_sync_module.os.fsync + fsync_calls = 0 + + def fail_directory_fsync(descriptor): + nonlocal fsync_calls + fsync_calls += 1 + if fsync_calls == 1: + raise OSError("directory fsync failed") + return real_fsync(descriptor) + + def fail_journal_recreation(path, flags, mode=0o777): + if os.path.abspath(path) == os.path.abspath(journal_path) and flags & os.O_CREAT: + raise OSError("journal recreation failed") + return real_open(path, flags, mode) + + try: + with ( + patch.object(sandbox_sync_module.os, "fsync", side_effect=fail_directory_fsync), + patch.object(sandbox_sync_module.os, "open", side_effect=fail_journal_recreation), + ): + with pytest.raises(OSError, match="directory fsync failed"): + sandbox_sync_module._unlink_journal_durably(journal_path) + + assert not journal_path.exists() + sandbox_sync_module._unlink_journal_durably(journal_path) + assert cleanup_stale_upload_staging_files(tmp_path) == 0 + replacement = publish_upload_bytes(uploads, "report.pdf", b"new primary") + assert replacement.name == "report_1.pdf" + finally: + upload_manager_module._release_uncertain_remote_journal_reservation(journal_path) + + assert cleanup_stale_upload_staging_files(tmp_path) == 1 + + @pytest.mark.parametrize("unlink_first", [False, True], ids=["unlink", "directory-fsync"]) def test_compensated_remote_delete_restores_host_even_when_journal_abort_fails(tmp_path, unlink_first): import deerflow.uploads.sandbox_sync as sandbox_sync_module @@ -272,7 +455,7 @@ def test_pending_remote_deletion_reserves_name_until_exact_remote_generation_is_ def update_file(self, _path: str, _content: bytes) -> None: raise OSError("compensation unavailable") - class Provider: + class Provider(_ExactProviderMixin): uses_thread_data_mounts = False def __init__(self, sandbox) -> None: @@ -346,7 +529,7 @@ def test_reconciliation_never_redirects_old_journal_to_new_sandbox(tmp_path): def remove_file(self, path: str) -> None: self.files.pop(path, None) - class RollingProvider: + class RollingProvider(_ExactProviderMixin): uses_thread_data_mounts = False def __init__(self, new_sandbox) -> None: @@ -472,14 +655,17 @@ def test_reconciliation_never_crosses_provider_boundary_when_raw_id_matches(tmp_ def test_reconciliation_finalizes_when_provider_confirms_exact_sandbox_absent(tmp_path): conversion_virtual = conversion_virtual_path("report.pdf") - class Provider: + class Provider(_ExactProviderMixin): def reconnect_sandbox_for_reconciliation( self, sandbox_id: str, *, thread_id: str, user_id: str | None, + identity, ): + assert identity.backend_namespace == "tests.backend" + assert identity.incarnation_id == "tests.incarnation" assert (sandbox_id, thread_id, user_id) == ("sandbox-old", "thread-1", "alice") from deerflow.sandbox.sandbox_provider import SandboxReconciliationResult @@ -546,7 +732,7 @@ def test_reconciliation_rejects_journal_that_does_not_match_retained_generation( def update_file(self, _path: str, _content: bytes) -> None: raise OSError("compensation unavailable") - class Provider: + class Provider(_ExactProviderMixin): uses_thread_data_mounts = False def __init__(self, sandbox) -> None: @@ -615,7 +801,7 @@ def test_reconciliation_accepts_exact_legacy_posix_filename(tmp_path, filename): def update_file(self, _path: str, _content: bytes) -> None: raise OSError("legacy compensation unavailable") - class Provider: + class Provider(_ExactProviderMixin): uses_thread_data_mounts = False def __init__(self, sandbox) -> None: @@ -721,7 +907,7 @@ def test_reconciliation_finishes_crash_after_first_remote_delete(tmp_path): def remove_file(self, path: str) -> None: self.files.pop(path, None) - class Provider: + class Provider(_ExactProviderMixin): uses_thread_data_mounts = False def __init__(self, sandbox) -> None: @@ -804,6 +990,44 @@ def test_prepare_deletion_rechecks_mount_mode_after_acquire(): assert provider.get_calls == 0 +def test_prepare_deletion_fails_before_mutation_without_restart_safe_identity(): + class Sandbox: + def __init__(self) -> None: + self.remove_calls: list[str] = [] + self.update_calls: list[str] = [] + + def remove_file(self, path: str) -> None: + self.remove_calls.append(path) + + def update_file(self, path: str, _content: bytes) -> None: + self.update_calls.append(path) + + class Provider: + uses_thread_data_mounts = False + + def __init__(self) -> None: + self.sandbox = Sandbox() + + def refresh_thread_data_mount_capabilities(self) -> bool: + return False + + def acquire(self, thread_id: str, *, user_id: str | None = None) -> str: + assert (thread_id, user_id) == ("thread-1", "alice") + return "sandbox-1" + + def get(self, sandbox_id: str): + assert sandbox_id == "sandbox-1" + return self.sandbox + + provider = Provider() + + with pytest.raises(RuntimeError, match="restart-safe backend and incarnation identity"): + prepare_upload_deletion(provider, "thread-1", user_id="alice") + + assert provider.sandbox.remove_calls == [] + assert provider.sandbox.update_calls == [] + + def test_prepare_deletion_reconciles_and_cleans_before_mount_decision(): import deerflow.uploads.sandbox_sync as sandbox_sync_module diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index d5cbdcbd8..4655d6ae7 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -16,6 +16,7 @@ from fastapi.testclient import TestClient from app.gateway.deps import get_config from app.gateway.routers import uploads from deerflow.sandbox.sandbox import Sandbox +from deerflow.sandbox.sandbox_provider import SandboxReconciliationIdentity, SandboxReconciliationResult from deerflow.uploads.layout import conversion_path_for_upload from deerflow.uploads.lease import UploadNameLease from deerflow.uploads.manager import create_upload_staging_file, delete_file_safe, publish_upload_bytes_leased @@ -42,6 +43,49 @@ def _mounted_provider() -> MagicMock: return provider +class _ExplicitSyncProvider: + uses_thread_data_mounts = False + + def __init__(self, sandbox) -> None: + self.sandbox = sandbox + + @staticmethod + def refresh_thread_data_mount_capabilities() -> bool: + return False + + async def acquire_async(self, _thread_id: str, *, user_id: str | None = None) -> str: + return "remote-1" + + def get(self, sandbox_id: str): + return self.sandbox if sandbox_id == "remote-1" else None + + @staticmethod + def reconciliation_provider_key() -> str: + return "tests.ExplicitSyncProvider" + + def prepare_sandbox_reconciliation_identity(self, sandbox_id: str) -> SandboxReconciliationIdentity: + assert sandbox_id == "remote-1" + return SandboxReconciliationIdentity( + provider_key=self.reconciliation_provider_key(), + backend_namespace="tests.backend", + incarnation_id="tests.incarnation", + ) + + def reconnect_sandbox_for_reconciliation( + self, + sandbox_id: str, + *, + thread_id: str, + user_id: str | None, + identity: SandboxReconciliationIdentity, + ) -> SandboxReconciliationResult: + del thread_id, user_id + expected = self.prepare_sandbox_reconciliation_identity(sandbox_id) + if identity != expected: + return SandboxReconciliationResult.unknown() + return SandboxReconciliationResult.found(self.sandbox) + + def _symlink_to_or_skip(link_path: Path, target_path: Path) -> None: try: link_path.symlink_to(target_path) @@ -1379,11 +1423,8 @@ def test_delete_uploaded_file_removes_explicitly_synced_remote_paths(tmp_path): conversion = conversion_path_for_upload(primary) conversion.parent.mkdir() conversion.write_text("generated", encoding="utf-8") - provider = MagicMock() - provider.uses_thread_data_mounts = False - provider.acquire_async = AsyncMock(return_value="remote-1") sandbox = MagicMock() - provider.get.return_value = sandbox + provider = _ExplicitSyncProvider(sandbox) with ( patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), @@ -1412,9 +1453,6 @@ def test_delete_uploaded_file_preserves_host_when_remote_delete_fails(tmp_path): thread_uploads_dir.mkdir(parents=True) primary = thread_uploads_dir / "notes.txt" primary.write_bytes(b"notes") - provider = MagicMock() - provider.uses_thread_data_mounts = False - provider.acquire_async = AsyncMock(return_value="remote-1") sandbox = MagicMock() def fail_remote_delete(_virtual_path): @@ -1422,7 +1460,7 @@ def test_delete_uploaded_file_preserves_host_when_remote_delete_fails(tmp_path): raise OSError("remote unavailable") sandbox.remove_file.side_effect = fail_remote_delete - provider.get.return_value = sandbox + provider = _ExplicitSyncProvider(sandbox) with ( patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), @@ -1450,12 +1488,9 @@ def test_delete_uploaded_file_compensates_partial_remote_delete_before_restoring conversion = conversion_path_for_upload(primary) conversion.parent.mkdir() conversion.write_text("generated", encoding="utf-8") - provider = MagicMock() - provider.uses_thread_data_mounts = False - provider.acquire_async = AsyncMock(return_value="remote-1") sandbox = MagicMock() sandbox.remove_file.side_effect = [None, OSError("conversion unavailable")] - provider.get.return_value = sandbox + provider = _ExplicitSyncProvider(sandbox) with ( patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), @@ -1488,13 +1523,10 @@ def test_delete_uploaded_file_commits_host_delete_when_remote_compensation_fails conversion = conversion_path_for_upload(primary) conversion.parent.mkdir() conversion.write_text("generated", encoding="utf-8") - provider = MagicMock() - provider.uses_thread_data_mounts = False - provider.acquire_async = AsyncMock(return_value="remote-1") sandbox = MagicMock() sandbox.remove_file.side_effect = [None, OSError("conversion unavailable")] sandbox.update_file.side_effect = OSError("remote rollback unavailable") - provider.get.return_value = sandbox + provider = _ExplicitSyncProvider(sandbox) with ( patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir), diff --git a/docker/provisioner/README.md b/docker/provisioner/README.md index fa03a8c98..f93442699 100644 --- a/docker/provisioner/README.md +++ b/docker/provisioner/README.md @@ -128,6 +128,29 @@ Get status and URL of a specific sandbox. **Status Values**: `Pending`, `Running`, `Succeeded`, `Failed`, `Unknown`, `NotFound` +### `GET /api/reconciliation/sandboxes/{sandbox_id}` + +Return the Kubernetes Namespace UID and the exact Pod UID used by durable +Gateway cleanup. This endpoint is Pod-authoritative: a missing Service with a +live Pod returns `status: "found"` and `sandbox_url: null`; only a missing Pod in +the same Namespace UID returns `status: "absent"`. The Gateway therefore cannot +redirect an old deletion journal to a replacement Pod that reused the same +deterministic `sandbox_id`. + +**Response**: +```json +{ + "sandbox_id": "abc-123", + "status": "found", + "backend_namespace": "namespace-uid", + "incarnation_id": "pod-uid", + "sandbox_url": "http://host.docker.internal:32123", + "user_id": "user-789", + "thread_id": "thread-456", + "mount_contract_version": 2 +} +``` + ### `DELETE /api/sandboxes/{sandbox_id}` Destroy a sandbox Pod + Service. diff --git a/docker/provisioner/app.py b/docker/provisioner/app.py index 070c99e1e..288670a36 100644 --- a/docker/provisioner/app.py +++ b/docker/provisioner/app.py @@ -11,6 +11,8 @@ run in K8s and are accessed by the backend via the configured Service mode. Endpoints: POST /api/sandboxes — Create a sandbox Pod + Service DELETE /api/sandboxes/{sandbox_id} — Destroy a sandbox Pod + Service + GET /api/reconciliation/sandboxes/{sandbox_id} + — Get Namespace/Pod immutable identity GET /api/sandboxes/{sandbox_id} — Get sandbox status & URL GET /api/sandboxes — List all sandboxes GET /health — Provisioner health check @@ -38,6 +40,7 @@ import re import secrets import time from contextlib import asynccontextmanager +from typing import Literal import urllib3 from fastapi import FastAPI, HTTPException, Request, Response @@ -526,6 +529,17 @@ class SandboxResponse(BaseModel): mount_contract_version: int | None = None +class SandboxReconciliationResponse(BaseModel): + sandbox_id: str + status: Literal["found", "absent"] + backend_namespace: str + incarnation_id: str | None = None + sandbox_url: str | None = None + user_id: str | None = None + thread_id: str | None = None + mount_contract_version: int | None = None + + # ── K8s resource helpers ───────────────────────────────────────────────── @@ -615,18 +629,21 @@ def _set_mount_contract_metadata( def _validate_existing_sandbox_contract( sandbox_id: str, *, + existing_pod: k8s_client.V1Pod | None = None, expected_pod: k8s_client.V1Pod | None = None, user_id: str | None = None, thread_id: str | None = None, ) -> tuple[str, str]: """Fail closed unless an existing Pod proves its tenant and mount contract.""" - try: - pod = core_v1.read_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE) - except ApiException as exc: - raise HTTPException( - status_code=409, - detail=f"Existing sandbox '{sandbox_id}' has no verifiable Pod contract", - ) from exc + pod = existing_pod + if pod is None: + try: + pod = core_v1.read_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE) + except ApiException as exc: + raise HTTPException( + status_code=409, + detail=f"Existing sandbox '{sandbox_id}' has no verifiable Pod contract", + ) from exc labels = pod.metadata.labels or {} annotations = pod.metadata.annotations or {} @@ -1476,6 +1493,68 @@ def destroy_sandbox(sandbox_id: str): return {"ok": True, "sandbox_id": sandbox_id} +@app.get( + "/api/reconciliation/sandboxes/{sandbox_id}", + response_model=SandboxReconciliationResponse, +) +def reconcile_sandbox_identity(sandbox_id: str): + """Return Pod-authoritative backend and immutable instance identity. + + The ordinary GET is access-oriented and returns 404 when its Service is + missing. Durable reconciliation must instead distinguish that partial + cleanup state from true Pod absence and must bind a deterministic raw ID to + the exact Pod UID that existed when the journal was prepared. + """ + try: + namespace = core_v1.read_namespace(K8S_NAMESPACE) + backend_namespace = getattr(namespace.metadata, "uid", None) + except ApiException as exc: + raise HTTPException( + status_code=503, + detail=f"Could not verify sandbox backend namespace: {exc.reason}", + ) from exc + if not isinstance(backend_namespace, str) or not backend_namespace: + raise HTTPException( + status_code=503, + detail="Sandbox backend namespace has no immutable UID", + ) + + try: + pod = core_v1.read_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE) + except ApiException as exc: + if exc.status == 404: + return SandboxReconciliationResponse( + sandbox_id=sandbox_id, + status="absent", + backend_namespace=backend_namespace, + ) + raise HTTPException( + status_code=503, + detail=f"Could not verify sandbox Pod: {exc.reason}", + ) from exc + + incarnation_id = getattr(pod.metadata, "uid", None) + if not isinstance(incarnation_id, str) or not incarnation_id: + raise HTTPException( + status_code=503, + detail=f"Sandbox '{sandbox_id}' Pod has no immutable UID", + ) + user_id, thread_id = _validate_existing_sandbox_contract( + sandbox_id, + existing_pod=pod, + ) + return SandboxReconciliationResponse( + sandbox_id=sandbox_id, + status="found", + backend_namespace=backend_namespace, + incarnation_id=incarnation_id, + sandbox_url=_sandbox_access_url(sandbox_id, tolerate_read_errors=True), + user_id=user_id, + thread_id=thread_id, + mount_contract_version=MOUNT_CONTRACT_VERSION, + ) + + @app.get("/api/sandboxes/{sandbox_id}", response_model=SandboxResponse) def get_sandbox(sandbox_id: str): """Return current status and URL for a sandbox."""