From 9379c634ed240d0225b63443606e5304a4e261c5 Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Fri, 14 Aug 2026 15:08:43 -0400 Subject: [PATCH] feat: scope backfill-by-copy and removal reconciliation jobs (#904) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: enforce explicit-document session purity in dedup/merge paths Audit for DEV-2000 (Scopes RFC prerequisite): explicit-level documents must stay session-pure so scope memory can be built by copying explicit documents between collections. Two classes of violation were possible: - Exact-content and semantic dedup in crud/document.py matched candidates with no level or session scoping, so an explicit document could be reinforced by — or soft-deleted in favor of — a same-content document from a different session or a different level (silently merging cross-session derivations into one row). - The generic create_observations tool handler accepted level='explicit' from agents with no message context (dreamer/dialectic), which would mint session-less explicit documents. Enforcement (refuse, never rewrite): - create_documents refuses explicit documents with a null session_name - exact dedup keys on (content, level, session-for-explicit); derived levels keep cross-session consolidation - is_rejected_duplicate scopes candidate search to the same level, and the same session for explicit documents - the create_observations tool rejects explicit-level input outside message ingestion (deriver) context Co-Authored-By: Claude Fable 5 * feat: add card_refresh dream type for event-driven peer-card updates Adds a lightweight dream variant (DEV-2000, Scopes RFC prerequisite) that runs ONLY the peer-card update — for event-driven refreshes such as scope membership changes and cold starts: - DreamType.CARD_REFRESH alongside OMNI; dispatched by process_dream to a new run_card_refresh_dream orchestration - CardRefreshSpecialist: restricted to get_recent_observations, search_memory, and update_peer_card (no observation-mutating tools), with a low tool-iteration cap of min(6, DREAM.MAX_TOOL_ITERATIONS) - rebuild=True mode carried in the dream payload: the existing card is NOT injected into the prompt and the specialist rebuilds it solely from observations present in the collection (for use after removals) - enqueue-able via the manual enqueue_dream path (bypasses volume gates); the work-unit key already embeds the dream type so a card refresh never collides with a pending omni dream. POST /v3/workspaces/{id}/schedule_dream accepts dream_type=card_refresh plus the rebuild flag - card refreshes never advance the omni dream guard pair (last_dream_at / last_dream_document_count) - shared PEER CARD prompt section extracted (verbatim) from DeductionSpecialist for reuse; CallPurpose gains dream.card_refresh Co-Authored-By: Claude Fable 5 * feat: reserve scope__ peer namespace with kind flag and guardrails Introduce the scope peer namespace (scope__) and the authoritative {"kind": "scope"} configuration flag, plus the server-side guardrails: - src/utils/scopes.py: single source of truth for the prefix, kind flag, and name helpers (scope_peer_name / is_scope_peer_name / scope_name_from_peer / validate_no_scope_peer_names) - reject reserved-prefix names on peer get-or-create (422) - reject scope peers as message authors in crud.create_messages (422) - reject scope peers as chat/representation targets (422); a scope peer as the path-level observer is deferred to Phase 2b - reject scope peers on the generic session-peer add/set/remove routes and the session-create peers mapping (422, directing to scopes routes) - peers.list excludes scope peers by default; new PeerGet.kind option ("scope" | "all") switches the view via a configuration JSONB filter - schemas: Scope / ScopeCreate / ScopeSessions(Add) and SessionCreate.scopes (unprefixed scope names, validated) Part of DEV-1997 (Scopes RFC DEV-1970). Co-Authored-By: Claude Fable 5 * feat: add scopes CRUD routes and session-create scopes wiring New /v3/workspaces/{workspace_id}/scopes facade (workspace-level auth; peer- and session-scoped keys are rejected): POST "" create-or-get (201/200) POST /list paginated scope list GET /{scope_id} single scope POST /{scope_id}/sessions add memberships DELETE /{scope_id}/sessions/{session_id} remove membership GET /{scope_id}/sessions list member session ids - crud/scope.py: get_or_create_scopes stamps the backing peer with {"kind": "scope", "observe_me": false} and refuses to adopt a legacy peer occupying the reserved name without the flag (409) - memberships are session_peers rows with observe_others=true / observe_me=false — identical to a hand-built observer peer - SessionCreate.scopes: create-or-get each scope peer and add the membership at session creation (the no-backfill common path) - crud/session.py: public upsert_session_peers wrapper so the facade bypasses the route-level guardrails without reaching into privates Backfill of pre-existing documents and reconciliation on removal land in DEV-1999; membership only affects messages ingested after the change. Part of DEV-1997 (Scopes RFC DEV-1970). Co-Authored-By: Claude Fable 5 * test: cover scopes facade, guardrails, and observer semantics - create-or-get idempotency, list/get, name validation, legacy-collision rejection (409), auth scoping (workspace key ok, peer/session keys 401) - reserved prefix rejected on peer create; peers.list kind filtering - scope peers rejected as message authors, chat/representation targets, and on the generic session-peer routes - membership add/list/remove with observe_others=true / observe_me=false row shape asserted via DB, and facade-less equivalence with a hand-built observer peer - end-to-end litmus: after adding a session to a scope, the deriver enqueue fan-out includes the scope peer as an observer - session creation with scopes: [a, b] creates both memberships Part of DEV-1997 (Scopes RFC DEV-1970). Co-Authored-By: Claude Fable 5 * feat: scope backfill-by-copy and removal reconciliation jobs Retroactive scope membership changes (DEV-1999): - New queue task types scope_backfill / scope_removal with payloads, work-unit keys ({task}:{workspace}:{scope_peer}:{session}), deduped enqueue (mirrors enqueue_dream), and consumer dispatch. - Backfill copies a session's explicit documents from each sender's global (P, P) collection into the scope's (scope_peer, P) collection with internal_metadata.copied_from as the idempotency marker; soft-deleted copies from an earlier removal are restored, so add -> remove -> re-add converges on exactly one live copy. Completion enqueues one manual omni dream per touched collection. - Removal soft-deletes the session's explicit documents in the scope's collections and cascades (fail-closed, transitively) to derived documents whose source_ids intersect anything removed, deletes the vectors from the external store, then enqueues a card_refresh dream with rebuild=True plus a manual omni dream per touched collection. - Zero LLM re-derivation: explicit documents are session-pure (DEV-2000 invariant); the only external call is re-embedding rows whose embedding column is NULL (external-store deployments). - Per-session job status lives in the scope peer's internal_metadata under backfill_status, written via single-statement JSONB merges (concurrent-writer safe) and surfaced at GET /v3/workspaces/{w}/scopes/{scope_id}/status. - Enqueued from the scopes add-sessions route and SessionCreate.scopes handling, only when the session already has messages. Co-Authored-By: Claude Fable 5 * test: add scope backfill/removal test coverage, fix backfill status JSONB bug Adds tests/deriver/test_scope_backfill.py covering the DEV-1999 scope backfill-by-copy and removal reconciliation jobs implemented in a prior commit: explicit-doc copying, copied_from idempotency (including add->remove->re-add), multi-peer collection routing, removal cascade to dependent derived docs, dream enqueues (manual omni on backfill; card_refresh rebuild + omni on removal), the status route, and add-sessions route wiring (backfill enqueued only when the session already has messages). Fixes a real production bug surfaced by these tests: update_scope_backfill_status passed json.dumps()'d strings through SQLAlchemy cast(..., JSONB), which double-encodes (psycopg re-serializes the already-JSON string), producing a JSONB string scalar instead of an object. Postgres's `||` between two non-array jsonb scalars doesn't merge — it silently wraps both into a 2-element array, corrupting backfill_status into a list. This crashed clear_scope_backfill_status's `#-` path delete (called on every removal) with "path element is not an integer" once a session had ever completed a backfill. Fixed by passing raw Python dicts to cast() instead, mirroring the working pattern already used in update_collection_internal_metadata. Also adds src.deriver.scope_backfill.tracked_db to conftest's tracked_db patch list — the module was missing from that per-import-site allowlist, so its DB work ran against the real configured database instead of the isolated per-test database. Co-Authored-By: Claude Fable 5 * fix(crud): preserve cache invalidation across get_or_create retry `get_or_create_peers` and `get_or_create_scopes` mutate existing rows, then insert new ones inside `db.begin_nested()`. When a concurrent writer creates one of those rows first, the insert raises IntegrityError and the function retries. `begin_nested()` autoflushes the pending UPDATEs *before* opening the savepoint, so the rollback neither undoes them nor expires the now-clean ORM state. The retry then compared already-updated values, found no change, and dropped those peers from `changed_peers` — skipping the cache purge while the row change committed anyway, leaving entries stale until the 300s TTL. Carry the mutated names into the retry via `_pending_invalidation` so the purge cannot be lost. The scopes facade mirrors `get_or_create_peers`, so both copies carried this. The peer path is pre-existing and runs on every message ingest. Also add the missing /v3 prefix to `_SCOPES_ROUTE_GUIDANCE`, which pointed callers at a 404. Adds tests/crud/test_get_or_create_retry_invalidation.py, which drives a real racing session and fails without this change. * fix(scopes): make scope identity unforgeable, unblock non-pattern peer names Three coupled changes to the scopes facade. 1. `PeerCreate` no longer gates internal lookups. It exists to validate a new, user-supplied peer id at the API boundary, but crud used it as a DTO for names that already exist, so any name outside RESOURCE_NAME_PATTERN raised a raw pydantic ValidationError — which is not a HonchoException, so it fell through to the catch-all handler as an HTTP 500. Adds `PeerSpec` (same fields, no charset pattern) as `PeerCreate`'s base, widens `get_or_create_peers` to accept it, and changes `get_peer` to take a plain str. All 13 construction sites converted; the create route keeps full validation. This unbreaks the Dreamer: DreamScheduler passes `collection.observer` straight into the specialist preflight, and scope peers have `observe_others=true`, so every `(scope.x, peer)` dream died there — the feature scopes exist to enable. It also fixes a pre-existing bug unrelated to scopes: a peer named `alice.smith` (legal before d429de0e5338, which validated names by length alone) 500s on message create, session peer add, and peer update. 2. The `kind` flag moves from `configuration` to `internal_metadata`. `configuration` is user-writable — `PeerCreate`/`PeerUpdate` accept a free-form dict and `update_peer` replaces it wholesale — so a legitimate `{"observe_me": true}` update silently dropped the flag, and a forged `{"kind": "scope"}` injected an ordinary peer into `POST /scopes/list`. `internal_metadata` appears in no API schema. `observe_me: false` stays in `configuration`, where it belongs. 3. Scope identity requires prefix AND flag, via `is_scope_peer()` and `scope_peer_clause()`. Neither half is forgeable: the prefix sits outside RESOURCE_NAME_PATTERN, `internal_metadata` is unreachable. Usage-site guards become flag-based so a legacy peer merely occupying the namespace keeps working rather than 422-ing on its own traffic; peer create and update stay name-based, since those must stop new names entering the namespace. `update_peer` now returns 422 instead of 500. Also swaps the reserved prefix from `scope__` to `scope.`: `_` is inside RESOURCE_NAME_PATTERN, so any tenant could already own a `scope__x` peer. No DB migration — `internal_metadata` already exists on `peers`, and no scope peers exist in any deployment yet. Co-Authored-By: Claude Opus 5 (1M context) * fix(scopes): validate peer names on create, close namespace squatting and upsert race Addresses three review findings against 48047a6a. 1. `PeerSpec` let API callers create invalid and reserved-prefix peers. Widening `get_or_create_peers` to accept a pattern-free schema fixed the lookup 500s but also removed validation from the *insert* path, and request-controlled names reach it via message authors, session peer maps, and the chat observer path — none of which carry a charset pattern of their own. Confirmed: `POST /sessions/{id}/messages` with `peer_id: "scope.x"` returned 201 and minted an unflagged squatter, after which `POST /scopes {id: x}` was permanently 409-blocked — namespace denial of service by any caller able to post a message. `peer_id: "not a valid name!@#"` was likewise created. Fixed by validating only names about to be INSERTed (`_validate_new_peer_names`), so already-existing names — legacy dotted names, scope peers — still resolve without a spurious 422. That keeps the Dreamer fix intact, since it reads through `get_peer`. 2. Existing reserved-prefix squatters could not be updated. The name-based guard on `PUT /peers/{peer_id}` refused every `scope.` name, contradicting the invariant that an unflagged squatter stays a normal peer. Now flag-based, so behavior is three-way: a real scope is refused, an existing unflagged peer updates, and a missing reserved-prefix name is refused by (1) rather than minted. 3. Scope checks raced with get-or-create and the membership upsert. The route-level guards run before peers are resolved, so a scope created concurrently in that window would be attached by the generic path with a default `SessionPeerConfig()`, clobbering its observer membership config. Adds `_reject_resolved_scope_peers`, which runs on the resolved rows in the same transaction as the upsert — no window, no extra query. The early checks stay for better error messages. Co-Authored-By: Claude Opus 5 (1M context) * fix(scopes): guard scope membership config, move checks to the mutation point Addresses a second review pass against 10655792. 1. Scope membership configuration was directly user-mutable. `PUT /sessions/{id}/peers/{peer_id}/config` had no scope guard at all, and `crud.set_peer_config` resolved the peer only to discard the row. Confirmed: posting `{"observe_others": false, "observe_me": true}` for a scope returned 204 and persisted, which silently stops all fan-out into the scope and makes Honcho form a representation *of* a scope — neither of which is reachable through the facade. Deterministic, no race required. Now checked on the row `get_peer` already returns, so it costs nothing and cannot race. 2. Empty and over-long names were still 500s. Removing the charset pattern from `PeerSpec` fixed one trap but left its length bounds, and request-bound peer names carry no length limits of their own — so `peer_id: ""` or a 513-char name reached `PeerSpec(...)` and raised a raw pydantic ValidationError that the catch-all turned into a 500. `PeerSpec` now carries no constraints at all (matching its documented purpose) and every rule for a new name lives in `_validate_new_peer_names` on the insert path. 3. Resolved-row protection generalized. The previous pass applied it only to membership upserts, leaving check-then-use windows elsewhere: peer update could have a concurrently-created scope's configuration replaced wholesale (create-path validation does not fire for a peer that now exists), the chat observer get-or-create could resolve a fresh scope as its observer, and the generic session-peer removal could silently detach a scope from its sessions. Each now inspects the resolved peer immediately before acting; the redundant name-level guard on the update route is dropped in favor of the race-free one. `remove_peers_from_session` grows an internal `_allow_scope_peers` flag because the scopes facade ends membership through that same path and must not be blocked by its own guard. Co-Authored-By: Claude Opus 5 (1M context) * test(scopes): enumerate every peer-touching route against a scope policy Three review passes each found the same class of defect: a route nobody had checked, rather than logic that was subtly wrong. One of them — `PUT /sessions/{id}/peers/{id}/config`, which let any caller set a scope to `observe_others=false` and silently stop all fan-out into it — predated this work entirely, because the guardrail set was assembled guardrail-by-guardrail instead of derived from the route list. Sampling review cannot close that kind of gap; enumeration can. Derives every route through which a peer name can reach the system and requires each to be classified as GUARDED or EXEMPT-with-a-reason, so a newly added peer-touching route fails the suite until someone classifies it. Detection is the union of path shape and a walk of the dependant tree (including sub-dependency `Form(...)` params and nested request-body models), because neither signal alone suffices: parameter names miss `POST /sessions/{id}/peers`, whose peer names are dict keys, and path shape misses `messages/upload`, whose `peer_id` arrives as a form field behind a parser dependency. Both invariants are then asserted behaviorally, by calling the routes rather than inspecting annotations — the guards deliberately live in crud, which is what makes `messages/upload` guarded for free via `crud.create_messages`: - a real scope is refused on all 11 guarded routes, and the rejection must name the scope, so an unrelated 422 (a malformed body) cannot pass the assertion; - an *unflagged* peer merely occupying the reserved namespace is unaffected. That half regressed once already when `update_peer` used a name-based check. Mutation-tested all three failure modes: disabling the `set_peer_config` guard fails the guarded test naming that route; regressing `update_peer` to name-based fails the squatter test; adding an unclassified peer route fails the enumeration. Covers the HTTP surface only. Peer names also reach the system through the deriver, dreamer, and queue, which have no route table to enumerate — noted in the module docstring rather than implied to be covered. Co-Authored-By: Claude Opus 5 (1M context) * fix(scopes): refuse a scope in every observed position; enumerate per position Addresses a fourth review pass against 62681e4d. The headline finding is that the previous commit's enumeration test had the wrong *model*, not a missing entry. 1. Manual conclusions could create knowledge about a scope. `POST /conclusions` validated only that observer_id and observed_id exist, so a scope as `observed_id` persisted a conclusion about a peer carrying observe_me=false and created an (observer, scope) collection for it. Confirmed: 201, and it read back. `POST /schedule_dream` had the same hole via `observed`. The fix is positional, because the invariant is: A scope may be an OBSERVER. A scope may never be OBSERVED. A scope as `observer_id` is how scoped conclusions are stored and must keep working (verified still 201); as `observed_id` it is now refused. Same split applied to schedule_dream `observed`, the peer-card `target` (which also covers a scope's self-card, since target-omitted collapses observed to peer_id), and session-context `peer_target`. 2. Chat target and both representation roles kept check-to-use races. Only the chat path-level observer was re-checked on its resolved row; the target was checked by name and then resolved without inspecting scope identity. Both are now checked at the dialectic preflight, where observer and observed are already resolved — an absent name has already failed by then, and an existing squatter cannot retroactively become a scope. 3. Generic membership removal was still racy. The adjacent SELECT narrowed the window but could not close it under READ COMMITTED. The UPDATE now carries its own correlated NOT EXISTS against scope_peer_clause(), so Postgres evaluates the exclusion as part of the statement and a scope committed after the advisory check still cannot be detached. 4. New-name validation ran after the name reached Postgres. A NUL byte passed the request schemas and PeerSpec, then raised psycopg.DataError inside the lookup — a 500. Values that cannot correspond to a stored row by construction (NUL bytes, over-length names) are now refused before the query. (Over-length names already returned 422; only the wasted query was real there.) The enumeration test is rekeyed from (method, path) to (method, path, position). A binary per-route verdict cannot express finding 1 at all: `POST /conclusions` is one route with two positions and opposite verdicts. Detection widens to observer / observed / target / peer_target / peer_perspective, which surfaced four routes the previous version never saw — conclusions, schedule_dream, queue/status, and session context. Also registers `src.routers.workspaces.tracked_db` in the conftest patch list; the new guard there would otherwise have run against the real configured database instead of the per-test one. Mutation-tested: disabling the conclusions observed-guard fails the positional test naming that position; adding an unclassified `observed_id` param fails enumeration. Co-Authored-By: Claude Opus 5 (1M context) * fix(scopes): refuse future scopes in observed positions, preserve scope membership Addresses a fifth review pass against 14136e5b. All six findings reproduced locally before fixing. 1. High — generic peer replacement removed scope memberships. `set_peers_for_session` soft-deleted every active SessionPeer row, and the request-level guard only inspected names *present* in the replacement map. A caller detached a scope by simply omitting it, never naming it — so no request-level guard could ever see it. Reproduced: scope sessions went [''] -> [] on a 200. The exclusion now lives in the UPDATE itself (correlated NOT EXISTS against scope_peer_clause), so replacement means "replace ordinary peers" regardless of request contents or concurrent creation. 2. High — peer cards could be pre-seeded for future scopes. `set_peer_card` resolves only the observer and writes a JSONB key derived from an unchecked observed name, and the route guard rejected only *existing* flagged scopes. Reproduced: PUT card with target=scope. returned 200, creating that scope then returned 201, and the card described the real scope. 3. High — dreams could be queued for future scopes. The route checked `observed` in a read-only session that closed before `enqueue_dream`, and a missing reserved name passes any is-it-a-scope check. Reproduced: 204 with observed=scope.. 2 and 3 share a root cause, so they share a fix: a new `reject_scope_observed` that is stricter than `reject_scope_peers` in exactly one case — a *missing* reserved name is refused, because nothing on these paths creates the peer, so nothing else would ever catch it. Existing unflagged squatters still pass. Both guards moved to the mutation point: card validation into `crud.set_peer_card` (same transaction as the JSONB write, so Dreamer and agent-tool callers are covered), dream validation into `enqueue_dream` (same transaction as the queue insert). The redundant route-level checks are dropped rather than left as weaker duplicates. 4. Medium — prefixed NUL names still reached PostgreSQL. `reject_scope_peers` filtered for the reserved prefix and sent matches to a text comparison, so "scope.future\0name" raised psycopg.DataError — a 500. Both guards now share `_reserved_name_candidates`, which materializes the input once and rejects impossible values before any SQL. Materializing matters independently: the message-author path passes a generator, and validation iterates separately from the prefix filter, so a generator would be half-consumed. `_reject_impossible_peer_names` now takes a Collection so the type checker enforces that. 5. Medium — representation kept a check-to-use race. The previous commit claimed both representation roles were rechecked after resolution; that was wrong — only the dialectic preflight got that check, and the representation route never goes through it. It now opens one short read-only session *after* the embedding call, checks both positions, and passes that same session to `get_working_representation`, so no connection is held across external work and a scope committed later cannot have conclusions in the collection being read. 6. Low — policy coverage was not exhaustive. `sender_id` reaches CRUD as `observed` but was missing from the detected parameter set. ALLOW cases could also not carry builders, so the suite never proved the other half of the contract — that legitimate scope *observers* keep working, which a guard rejecting scopes everywhere would satisfy. Both fixed; observer positions on conclusions, dreams, cards, session context and queue status are now asserted behaviorally. Deliberately not implemented: the scope-creation backstop scanning for pre-existing card keys and queue items naming a future backing peer. Reasoning is recorded in `get_or_create_scopes` — no new such state can be created now, any pre-existing row is coincidental since `scope.` was never a meaningful namespace, the consequence is inert, and detecting card keys means a full table scan per scope creation. Mutation-tested each new guard: removing the replacement exclusion fails both membership-preservation tests; weakening either observed guard to existing-only fails the pre-seeding tests. Co-Authored-By: Claude Opus 5 (1M context) * fix(scopes): exclude scope memberships from the session observer limit Scope memberships carry observe_others=true, so every scope counted against SESSION_OBSERVERS_LIMIT (default 10) — capping scopes-per-session at the limit minus the session's real observers, and reporting the failure as `400 Cannot create session with 11 observers. ... Observers are peers with 'observe_others' set to true.` on a membership call. Wrong on three counts: the ceiling is undocumented and contradicts RFC §5.1 ("sessions belong to any number of scopes"), the message describes session creation, and it leaks the word "observer" through a facade whose entire job is hiding observers (RFC goal 5). The limit exists to bound per-observer deriver fan-out for real peers; a scope costs document rows, not LLM calls (RFC §5.2), so it does not belong in that budget. Excluded from both halves of the check in `_get_or_add_peers_to_session`: the incoming names via a flag-based lookup, existing memberships via a correlated NOT EXISTS on `scope_peer_clause()` — the same pattern the replacement and removal paths already use, so the exclusion holds regardless of concurrent scope creation. The early `count_observers_in_config(session.peer_names)` check in `get_or_create_session` is left alone: `peer_names` cannot contain a scope, and `scopes` is a separate field. `reject_scope_peers` is split into a `scope_peer_names()` query helper plus a two-line raiser so the observer count reuses the authoritative name-AND-flag predicate instead of growing a third copy of it. Still costs nothing on the common path — no reserved-prefix name in the input means no query at all. Also caps `SessionCreate.scopes` at 100, matching `ScopeSessionsAdd.session_ids`. This belongs in the same commit: the observer limit was the only thing bounding that list, so removing it turns an unbounded `scopes` array into a peer row and a membership row per element, committed — the single-request path to the cardinality anti-pattern RFC §8 warns about. Partly answers OQ6: no per-session cap, 100 per request. Tests: a session joins SESSION_OBSERVERS_LIMIT + 2 scopes through both the facade and session creation; real observers over the limit still 400, so the carve-out cannot quietly disable the limit; 101 scopes is a 422. Co-Authored-By: Claude Opus 5 (1M context) * fix(scopes): skip semantic retrieval when the embedding precompute failed Addresses three open review comments. 1. Major — the representation read could embed inside its DB session. The route's precompute is suppressed, and both `RepresentationManager.get_working_representation` and `crud.query_documents` fall back to embedding when a query arrives without one, so a failed precompute meant an external call inside the read session this branch opens for the scope re-check — the connection-holding rule the route's own comment claimed to satisfy. The innermost fallback also only catches ValueError, so a provider outage surfaced as a 500. The semantic query is now passed only when an embedding exists, degrading to derived+recent retrieval. (`crud.query_documents` embedding inside a caller's session predates this branch and is left alone.) 2. Minor — `test_resolved_scope_peer_rejected_at_membership_upsert` described a race it does not perform. It creates an already-flagged scope and calls crud directly; the unflagged → flagged transition is not simulated. Docstring now says what the test actually pins. 3. Minor — `test_empty_replacement_preserves_scope_membership` asserted only half its docstring. It passed if the empty PUT left every ordinary membership intact; now asserts the ordinary peer's left_at is set. Co-Authored-By: Claude Opus 5 (1M context) * fix(scopes): close auth and observed-position gaps, paginate membership Review response for #884. Security: - gate `SessionCreate.scopes` behind a workspace-level key; the session-create route is self-authorizing, so a peer- or session-scoped token could mint scope peers and join sessions to scopes it had no access to via `POST /scopes` - refuse a reserved-but-nonexistent name in two observed positions that used the permissive guard: chat `target` and session-context `peer_target`. Both let a caller act on `scope.X` before it existed, then create the scope Facade: - exclude scope peers from `GET /sessions/{id}/peers` and refuse the membership -config read for a real scope, matching its write side - replace `GET /scopes/{id}/sessions` with `POST /scopes/{id}/sessions/list` returning `Page[Session]`; the add route now returns 204. Membership was unbounded on both, while every other list surface paginates - rename `crud.get_scope` to `get_scope_or_raise` Tests: - add a missing-name axis to the route-policy table (`Case.refuse_missing`), which is what surfaced the two guard gaps above - delete 14 hand-written tests the table now enumerates; 52 -> 39 functions in test_scopes.py with more cases covered - tighten the squatter assertion from `!= 422` to `< 400`, which was passing on 5xx - assert the FastAPI-internals traversal still derives positions, so a framework upgrade can't silently empty the suite Docs: - drop internal ticket and RFC references from the published OpenAPI descriptions and surrounding comments; state the behavior instead - move implementation reasoning out of the `PUT /peers/{id}` docstring, which FastAPI publishes, into a comment * test(scopes): assert exact statuses for permissive missing-name cases Follow-up review pass on #884. - add `Case.missing_status` so a permissive missing-name position asserts the status it should actually get (404, or 200 for the no-op removal) instead of `!= 422`, which also passed on a 5xx — the same hole already closed in the squatter assertion - require it whenever `refuse_missing` is False, and require its absence when True, so the policy table can't drift from the assertion - repoint a stale allow-reason at POST /scopes/{scope_id}/sessions/list; the GET it named was removed - document the membership list's ordering under `reverse` * chore: clean up stale docstring language * fix: don't backfill a session that left the scope scope_backfill and scope_removal carry different work-unit keys, so nothing orders them: a removal enqueued right after the add — or one that lands while the backfill is embedding — sweeps the scope before the copies exist, leaving a departed session's documents live in the scope forever. _run_backfill now re-checks SessionPeer membership inside the write transaction and returns None; process_scope_backfill then skips both the dream enqueues and the status write, so a skipped backfill can't resurrect the status entry removal just cleared. Adds coverage for the skip, the NULL-embedding re-embed path, and the failed-status write. Handler-driven tests now stand up the membership row the guard requires. Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Fable 5 --- src/crud/__init__.py | 8 + src/crud/scope.py | 178 ++++- src/crud/session.py | 28 +- src/deriver/consumer.py | 36 + src/deriver/enqueue.py | 163 +++++ src/deriver/scope_backfill.py | 558 ++++++++++++++++ src/routers/scopes.py | 39 +- src/schemas/__init__.py | 2 + src/schemas/api.py | 18 +- src/utils/queue_payload.py | 36 + src/utils/types.py | 9 +- src/utils/work_unit.py | 24 + tests/conftest.py | 1 + tests/deriver/test_scope_backfill.py | 958 +++++++++++++++++++++++++++ 14 files changed, 2036 insertions(+), 22 deletions(-) create mode 100644 src/deriver/scope_backfill.py create mode 100644 tests/deriver/test_scope_backfill.py diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 7ce4c249..38be14da 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -49,12 +49,16 @@ from .representation import ( ) from .scope import ( add_sessions_to_scope, + clear_scope_backfill_status, get_or_create_scopes, + get_scope_backfill_status, get_scope_or_raise, get_scope_sessions, get_scopes, + invalidate_scope_peer_cache, remove_session_from_scope, resolve_scope_peers, + update_scope_backfill_status, ) from .session import ( SessionDeletionResult, @@ -137,12 +141,16 @@ __all__ = [ "get_working_representation", # Scope "add_sessions_to_scope", + "clear_scope_backfill_status", "get_or_create_scopes", + "get_scope_backfill_status", "get_scope_or_raise", "get_scope_sessions", "get_scopes", + "invalidate_scope_peer_cache", "remove_session_from_scope", "resolve_scope_peers", + "update_scope_backfill_status", # Session "SessionDeletionResult", "get_sessions", diff --git a/src/crud/scope.py b/src/crud/scope.py index a2ef1588..d5b92c31 100644 --- a/src/crud/scope.py +++ b/src/crud/scope.py @@ -7,17 +7,26 @@ authoritative, user-unwritable flag) and ``{"observe_me": false}`` in and never speaks. See ``src/utils/scopes.py`` for the namespace helpers. -Membership only affects messages ingested *after* a session is added to a -scope. Conclusions already derived are neither backfilled on add nor -reconciled on removal. +Messages ingested after a membership change flow to the scope via the normal +deriver fan-out. Retroactive changes are handled by queue jobs: adding a session +that already has messages enqueues a ``scope_backfill`` task (copy of the +session's explicit documents into the scope's collections) and removing a +session enqueues a ``scope_removal`` task (soft-delete of the session's +documents plus dependent derived documents). """ from collections.abc import Sequence +from datetime import datetime, timezone from logging import getLogger +from typing import Any, Literal +from typing import cast as py_cast -from sqlalchemy import Select, select +from sqlalchemy import Select, Text, cast, select, update +from sqlalchemy.dialects.postgresql import ARRAY, JSONB, array +from sqlalchemy.engine import CursorResult from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.sql.functions import func from src import models, schemas from src.cache.client import safe_cache_delete @@ -38,11 +47,21 @@ from .workspace import get_or_create_workspace logger = getLogger(__name__) +# Key inside the scope peer's internal_metadata that holds per-session +# backfill job status: {: {state, updated_at[, docs_copied]}}. +BACKFILL_STATUS_KEY = "backfill_status" + +ScopeBackfillState = Literal["pending", "completed", "failed"] + # Internal metadata stamped on every scope peer at creation. `kind` is the # authoritative scope flag and lives here — NOT in `configuration` — because # `configuration` is user-writable (`PeerCreate`/`PeerUpdate` accept a free-form # dict, and `update_peer` replaces it wholesale), so a user could forge or clear # the flag. `internal_metadata` appears in no API schema at all. +# +# Backfill job status shares this field under BACKFILL_STATUS_KEY below. Every +# write to it must be a JSONB merge scoped to that key, never a wholesale +# replacement, or the `kind` flag goes with it and the peer stops being a scope. SCOPE_PEER_INTERNAL_METADATA: dict[str, str] = { "kind": SCOPE_KIND, } @@ -355,9 +374,9 @@ async def add_sessions_to_scope( Each membership is a ``session_peers`` row for the scope peer with ``observe_others=true, observe_me=false`` — exactly what a hand-built - observer peer would carry. No backfill happens here: membership only affects - messages ingested after this call, and conclusions already derived are left - as they are. + observer peer would carry. Sessions that already have messages get a + ``scope_backfill`` queue task so their existing explicit documents are + copied into the scope's collections; fresh sessions need nothing. Args: db: Database session @@ -400,6 +419,23 @@ async def add_sessions_to_scope( await db.commit() + # Backfill: only sessions that already have messages need it. + # Imported lazily: src.deriver.enqueue imports crud at module level. + from src.deriver.enqueue import enqueue_scope_backfill + + msg_result = await db.execute( + select(models.Message.session_name) + .where(models.Message.workspace_name == workspace_name) + .where(models.Message.session_name.in_(requested)) + .distinct() + ) + for session_with_messages in sorted({row[0] for row in msg_result.all()}): + await enqueue_scope_backfill( + workspace_name, + scope_peer=scope_peer_name(scope_name), + session_name=session_with_messages, + ) + async def remove_session_from_scope( db: AsyncSession, @@ -411,8 +447,9 @@ async def remove_session_from_scope( Remove a session from a scope by ending the scope peer's membership. Ends the membership the same way the generic remove-peer path does (sets - ``left_at``). Conclusions derived while the session was a member are left in - place — nothing reconciles them. + ``left_at``), then enqueues a ``scope_removal`` reconciliation task that + soft-deletes the session's documents from the scope's collections along + with any derived documents whose support left the scope. Args: db: Database session @@ -424,6 +461,8 @@ async def remove_session_from_scope( ResourceNotFoundException: If the scope or session does not exist """ # Lazy import for the same circular-import reason as add_sessions_to_scope. + from src.deriver.enqueue import enqueue_scope_removal + from .session import remove_peers_from_session await get_scope_or_raise(db, workspace_name, scope_name) @@ -436,3 +475,124 @@ async def remove_session_from_scope( # This *is* the supported path for ending scope membership. _allow_scope_peers=True, ) + + await enqueue_scope_removal( + workspace_name, + scope_peer=scope_peer_name(scope_name), + session_name=session_name, + ) + + +async def update_scope_backfill_status( + db: AsyncSession, + workspace_name: str, + scope_peer: str, + session_name: str, + *, + state: ScopeBackfillState, + docs_copied: int | None = None, +) -> None: + """ + Merge one session's backfill status into the scope peer's internal_metadata. + + Uses a single-statement nested JSONB merge so concurrent writers updating + *different* session keys never clobber each other (the merge is computed + from the row's current committed value under the row lock, never from a + stale Python-side read): + + internal_metadata || {"backfill_status": + coalesce(internal_metadata->'backfill_status', '{}') || {: }} + + Does not commit; the caller owns the transaction. Callers should + invalidate the peer cache after committing (``peer_cache_key``). + """ + entry: dict[str, Any] = { + "state": state, + "updated_at": datetime.now(timezone.utc).isoformat(), + } + if docs_copied is not None: + entry["docs_copied"] = docs_copied + + # NOTE: the JSONB operand must be a Python dict, not a json.dumps() string. + # cast(, JSONB) double-encodes: psycopg's JSONB + # bind adapter serializes the *string* again, producing a JSONB string + # scalar instead of an object. `||` between two non-array jsonb scalars + # doesn't merge keys — it silently wraps both sides into a 2-element + # array, corrupting backfill_status into a list and later crashing + # clear_scope_backfill_status's `#-` path delete (which then sees an + # array where it expects an object). + merged_status = func.coalesce( + models.Peer.internal_metadata.op("->")(BACKFILL_STATUS_KEY), + cast({}, JSONB), + ).op("||")(cast({session_name: entry}, JSONB)) + + stmt = ( + update(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name == scope_peer) + .values( + internal_metadata=models.Peer.internal_metadata.op("||")( + func.jsonb_build_object(BACKFILL_STATUS_KEY, merged_status) + ) + ) + ) + result = py_cast(CursorResult[Any], await db.execute(stmt)) + if result.rowcount == 0: + raise ResourceNotFoundException( + f"Scope peer {scope_peer} not found in workspace {workspace_name}" + ) + + +async def clear_scope_backfill_status( + db: AsyncSession, + workspace_name: str, + scope_peer: str, + session_name: str, +) -> None: + """ + Drop one session's entry from the scope peer's backfill status. + + Called by the removal reconciliation job: once a session leaves a scope its + backfill status is moot, and clearing it keeps add→remove→re-add cycles + honest (the re-add starts from a fresh ``pending``). Single-statement + ``#-`` delete, safe against concurrent per-key merges. Does not commit. + Missing peers are a no-op (removal is idempotent). + """ + stmt = ( + update(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name == scope_peer) + .values( + internal_metadata=models.Peer.internal_metadata.op("#-")( + cast(array([BACKFILL_STATUS_KEY, session_name]), ARRAY(Text)) + ) + ) + ) + await db.execute(stmt) + + +async def invalidate_scope_peer_cache(workspace_name: str, scope_peer: str) -> None: + """Invalidate the cached peer row after a status write commits.""" + await safe_cache_delete(peer_cache_key(workspace_name, scope_peer)) + + +async def get_scope_backfill_status( + db: AsyncSession, + workspace_name: str, + scope_name: str, +) -> dict[str, Any]: + """ + Read the per-session backfill job status map for a scope. + + Returns: + ``{: {state, updated_at[, docs_copied]}}`` (empty when no + backfill has ever been enqueued for the scope) + + Raises: + ResourceNotFoundException: If the scope does not exist + """ + peer = await get_scope_or_raise(db, workspace_name, scope_name) + status = peer.internal_metadata.get(BACKFILL_STATUS_KEY, {}) + if not isinstance(status, dict): + return {} + return py_cast(dict[str, Any], status) diff --git a/src/crud/session.py b/src/crud/session.py index 1fc884ec..efa3f664 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -320,8 +320,8 @@ async def get_or_create_session( # Add the session to any requested scopes: create-or-get each scope peer # and record an observer membership (observe_others=true, observe_me=false). - # No backfill happens here — membership only affects messages ingested - # after this point. + # If the session already has messages, a backfill task is enqueued after + # commit (below) so its existing documents are copied into the scope. scopes_result = None if session.scopes: scopes_result = await get_or_create_scopes( @@ -352,6 +352,30 @@ async def get_or_create_session( if scopes_result is not None: await scopes_result.post_commit() + # Backfill (DEV-1999): a pre-existing session added to scopes at + # create-or-get time may already have messages; those need a + # backfill-by-copy task per scope. Fresh sessions need nothing. + if session.scopes: + has_messages = await db.scalar( + select( + exists( + select(models.Message.id) + .where(models.Message.workspace_name == workspace_name) + .where(models.Message.session_name == session.name) + ) + ) + ) + if has_messages: + # Imported lazily: src.deriver.enqueue imports crud at module level. + from src.deriver.enqueue import enqueue_scope_backfill + + for scope_name in session.scopes: + await enqueue_scope_backfill( + workspace_name, + scope_peer=scope_peer_name(scope_name), + session_name=session.name, + ) + # Only update cache if session data changed or was newly created if needs_cache_update: cache_key = session_cache_key(workspace_name, session.name) diff --git a/src/deriver/consumer.py b/src/deriver/consumer.py index 708d05f6..118135ca 100644 --- a/src/deriver/consumer.py +++ b/src/deriver/consumer.py @@ -8,6 +8,10 @@ from sqlalchemy import select from src import crud, models from src.dependencies import tracked_db from src.deriver.deriver import process_representation_tasks_batch +from src.deriver.scope_backfill import ( + process_scope_backfill, + process_scope_removal, +) from src.dreamer import process_dream from src.exceptions import ResourceNotFoundException, ValidationException from src.models import Message @@ -26,6 +30,8 @@ from src.utils.queue_payload import ( DeletionPayload, DreamPayload, ReconcilerPayload, + ScopeBackfillPayload, + ScopeRemovalPayload, SummaryPayload, WebhookPayload, ) @@ -150,6 +156,36 @@ async def process_item(queue_item: models.QueueItem) -> None: raise ValueError(f"Invalid payload structure: {str(e)}") from e await process_deletion(validated, workspace_name) + elif task_type == "scope_backfill": + with sentry_sdk.start_transaction( + name="process_scope_backfill_task", op="deriver" + ): + try: + validated = ScopeBackfillPayload(**queue_payload) + except ValidationError as e: + logger.error( + "Invalid scope_backfill payload received: %s. Payload: %s", + str(e), + queue_payload, + ) + raise ValueError(f"Invalid payload structure: {str(e)}") from e + await process_scope_backfill(validated, workspace_name) + + elif task_type == "scope_removal": + with sentry_sdk.start_transaction( + name="process_scope_removal_task", op="deriver" + ): + try: + validated = ScopeRemovalPayload(**queue_payload) + except ValidationError as e: + logger.error( + "Invalid scope_removal payload received: %s. Payload: %s", + str(e), + queue_payload, + ) + raise ValueError(f"Invalid payload structure: {str(e)}") from e + await process_scope_removal(validated, workspace_name) + else: raise ValueError(f"Invalid task type: {task_type}") diff --git a/src/deriver/enqueue.py b/src/deriver/enqueue.py index f62a3db3..2f70e787 100644 --- a/src/deriver/enqueue.py +++ b/src/deriver/enqueue.py @@ -16,6 +16,7 @@ from src.utils.queue_payload import ( create_deletion_payload, create_dream_payload, create_payload, + create_scope_task_payload, ) from src.utils.work_unit import construct_work_unit_key @@ -569,6 +570,168 @@ async def enqueue_dream( raise +def create_scope_task_record( + workspace_name: str, + *, + task_type: Literal["scope_backfill", "scope_removal"], + scope_peer: str, + session_name: str, +) -> dict[str, Any]: + """ + Create a queue record for a scope backfill / removal task (DEV-1999). + + Args: + workspace_name: Name of the workspace + task_type: "scope_backfill" or "scope_removal" + scope_peer: Prefixed name of the peer backing the scope + session_name: Name of the session whose membership changed + + Returns: + Queue record dictionary ready for insertion into the queue + """ + payload = create_scope_task_payload( + task_type, scope_peer=scope_peer, session_name=session_name + ) + return { + "work_unit_key": construct_work_unit_key(workspace_name, payload), + "payload": payload, + "session_id": None, + "task_type": task_type, + "workspace_name": workspace_name, + "message_id": None, + } + + +async def _enqueue_scope_task( + workspace_name: str, + *, + task_type: Literal["scope_backfill", "scope_removal"], + scope_peer: str, + session_name: str, +) -> None: + """ + Enqueue a scope backfill / removal task, deduplicating like enqueue_dream. + + If a task with the same work_unit_key is already in progress (has an + ActiveQueueSession) or pending in the queue, the enqueue is skipped — the + handlers are idempotent, so a queued task already covers this membership + change. + + For backfill tasks, the scope peer's per-session status entry is written + to "pending" in the same transaction as the queue insert, so the status + surface never claims a job exists that was never enqueued (or vice versa). + """ + async with tracked_db("scope_task_enqueue") as db_session: + try: + record = create_scope_task_record( + workspace_name, + task_type=task_type, + scope_peer=scope_peer, + session_name=session_name, + ) + work_unit_key = record["work_unit_key"] + + is_in_progress = await db_session.scalar( + select( + exists( + select(models.ActiveQueueSession.id).where( + models.ActiveQueueSession.work_unit_key == work_unit_key + ) + ) + ) + ) + if is_in_progress: + logger.debug( + "Skipping %s enqueue - already in progress: %s", + task_type, + work_unit_key, + ) + return + + is_pending = await db_session.scalar( + select( + exists( + select(QueueItem.id).where( + QueueItem.work_unit_key == work_unit_key, + QueueItem.processed == False, # noqa: E712 + ) + ) + ) + ) + if is_pending: + logger.debug( + "%s already pending in queue: %s", task_type, work_unit_key + ) + return + + stmt = insert(QueueItem).returning(QueueItem) + await db_session.execute(stmt, [record]) + + if task_type == "scope_backfill": + await crud.update_scope_backfill_status( + db_session, + workspace_name, + scope_peer, + session_name, + state="pending", + ) + + await db_session.commit() + await crud.invalidate_scope_peer_cache(workspace_name, scope_peer) + + logger.info( + "Enqueued %s task for %s/%s/%s", + task_type, + workspace_name, + scope_peer, + session_name, + ) + + except Exception as e: + logger.exception("Failed to enqueue %s task!", task_type) + if settings.SENTRY.ENABLED: + import sentry_sdk + + sentry_sdk.capture_exception(e) + raise + + +async def enqueue_scope_backfill( + workspace_name: str, + *, + scope_peer: str, + session_name: str, +) -> None: + """ + Enqueue a backfill-by-copy task for a session newly added to a scope. + + Only call this for sessions that already have messages — fresh sessions + need nothing (the deriver fan-out covers everything ingested after the + membership change). + """ + await _enqueue_scope_task( + workspace_name, + task_type="scope_backfill", + scope_peer=scope_peer, + session_name=session_name, + ) + + +async def enqueue_scope_removal( + workspace_name: str, + *, + scope_peer: str, + session_name: str, +) -> None: + """Enqueue a removal reconciliation task for a session removed from a scope.""" + await _enqueue_scope_task( + workspace_name, + task_type="scope_removal", + scope_peer=scope_peer, + session_name=session_name, + ) + + def create_deletion_record( workspace_name: str, deletion_type: Literal["session", "observation", "workspace"], diff --git a/src/deriver/scope_backfill.py b/src/deriver/scope_backfill.py new file mode 100644 index 00000000..81017e28 --- /dev/null +++ b/src/deriver/scope_backfill.py @@ -0,0 +1,558 @@ +"""Scope membership reconciliation jobs (DEV-1999, part of the Scopes RFC). + +Two queue task handlers, dispatched from ``consumer.process_item``: + +- ``scope_backfill`` — a session was added to a scope that already had + messages. Copy the session's *explicit* documents from each sender peer's + global ``(P, P)`` collection into the scope's ``(scope_peer, P)`` collection, + then enqueue a manual omni dream per touched collection to rebuild the + scope's higher-order layer and card. +- ``scope_removal`` — a session was removed from a scope. Soft-delete the + session's explicit documents from the scope's collections, cascade the + soft-delete to derived documents whose support left the scope (fail-closed), + then enqueue a ``card_refresh`` dream with ``rebuild=True`` plus a manual + omni dream per touched collection. + +Zero LLM re-derivation: explicit-level documents are session-pure and +identical across observer collections (the DEV-2000 invariant), so retroactive +membership is pure row copying. The only external call is an embedding lookup +for source rows whose embedding column is NULL — embedding API only, never an +LLM. + +DB sessions are never held across embedding or vector-store calls: the +handlers run in phases (plan → embed → write → sync), each phase opening its +own short-lived session. +""" + +import logging +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import select, update +from sqlalchemy.dialects.postgresql import array +from sqlalchemy.sql.functions import func + +from src import crud, models +from src.config import settings +from src.crud.scope import ScopeBackfillState +from src.crud.session import is_peer_in_session +from src.dependencies import tracked_db +from src.embedding_client import embedding_client +from src.schemas import DreamType +from src.utils.queue_payload import ScopeBackfillPayload, ScopeRemovalPayload +from src.utils.scopes import is_scope_peer_name +from src.vector_store import VectorRecord, get_external_vector_store + +logger = logging.getLogger(__name__) + +# internal_metadata key linking a backfilled copy to the global document it +# was copied from. The presence of this key is the idempotency marker. +COPIED_FROM_KEY = "copied_from" + + +def _store_embeddings_in_postgres() -> bool: + """Whether document embeddings are persisted to the postgres column. + + True when TYPE=pgvector OR still migrating (dual-write) — mirrors + ``crud.document.create_documents``. + """ + return ( + settings.VECTOR_STORE.TYPE == "pgvector" or not settings.VECTOR_STORE.MIGRATED + ) + + +def _embedding_as_list(embedding: Any) -> list[float] | None: + """Normalize a pgvector column value (numpy array or None) to a list.""" + if embedding is None: + return None + return list(embedding) + + +@dataclass +class _CopySpec: + """A planned copy of one global explicit document into a scope collection. + + ``restore_document_id`` points at an existing soft-deleted copy to restore + when set; for new copies it is filled with the inserted row's id after the + write phase. + """ + + observed: str + source_id: str + content: str + embedding: list[float] | None + internal_metadata: dict[str, Any] + times_derived: int + source_ids: list[str] | None + session_name: str + restore_document_id: str | None = None + + +# --------------------------------------------------------------------------- +# Backfill (session added to a scope) +# --------------------------------------------------------------------------- + + +async def process_scope_backfill( + payload: ScopeBackfillPayload, workspace_name: str +) -> None: + """Process a ``scope_backfill`` queue task. + + Idempotent: a live copy (matched by ``internal_metadata.copied_from``) is + never duplicated, and a soft-deleted copy left by an earlier removal is + restored rather than re-inserted, so add→remove→re-add converges on + exactly one live copy per source document. + """ + scope_peer = payload.scope_peer + session_name = payload.session_name + + try: + result = await _run_backfill(workspace_name, scope_peer, session_name) + except Exception: + await _write_backfill_status( + workspace_name, scope_peer, session_name, state="failed" + ) + raise + + if result is None: + # The session left the scope before this task was drained: removal + # already cleared the copies and the status entry, so there is nothing + # to backfill and nothing to record. + logger.info( + "Scope backfill skipped for %s/%s/%s: membership is no longer active", + workspace_name, + scope_peer, + session_name, + ) + return + docs_copied, touched_observed = result + + # One manual (gate-bypassing) omni dream per touched collection builds the + # scope's higher-order layer and card. enqueue_dream dedupes on the + # work-unit key, so a batch of backfills collapses to one dream each. + from src.deriver.enqueue import enqueue_dream + + for observed in sorted(touched_observed): + await enqueue_dream( + workspace_name, + observer=scope_peer, + observed=observed, + dream_type=DreamType.OMNI, + trigger_reason="scope_backfill", + ) + + await _write_backfill_status( + workspace_name, + scope_peer, + session_name, + state="completed", + docs_copied=docs_copied, + ) + + logger.info( + "Scope backfill complete for %s/%s/%s: %d documents copied across %d collections", + workspace_name, + scope_peer, + session_name, + docs_copied, + len(touched_observed), + ) + + +async def _run_backfill( + workspace_name: str, scope_peer: str, session_name: str +) -> tuple[int, set[str]] | None: + """Run the copy itself. + + Returns (docs copied or restored, touched observed peers), or ``None`` if + the session is no longer a member of the scope. + """ + # Phase 1 (DB): plan. Read the session's explicit documents from each + # sender's global (P, P) collection, and any existing copies (live or + # soft-deleted) already in the scope's collections. + plans: list[_CopySpec] = [] + async with tracked_db("scope_backfill.plan") as db: + source_result = await db.execute( + select(models.Document).where( + models.Document.workspace_name == workspace_name, + models.Document.session_name == session_name, + models.Document.level == "explicit", + models.Document.observer == models.Document.observed, + models.Document.deleted_at.is_(None), + ) + ) + source_docs = [ + doc + for doc in source_result.scalars().all() + # Scope peers never speak and are never observed, but stay + # defensive: never treat another scope's rows as a source. + if not is_scope_peer_name(doc.observer) + ] + + if not source_docs: + return 0, set() + + # Existing copies in the scope's collections for this session, keyed + # by (observed, copied_from). Includes soft-deleted rows: those are + # restore candidates, not blockers. + copies_result = await db.execute( + select(models.Document).where( + models.Document.workspace_name == workspace_name, + models.Document.observer == scope_peer, + models.Document.session_name == session_name, + models.Document.internal_metadata.has_key(COPIED_FROM_KEY), + ) + ) + live_copies: set[tuple[str, str]] = set() + soft_deleted_copies: dict[tuple[str, str], str] = {} + for copy_doc in copies_result.scalars().all(): + key = (copy_doc.observed, str(copy_doc.internal_metadata[COPIED_FROM_KEY])) + if copy_doc.deleted_at is None: + live_copies.add(key) + else: + soft_deleted_copies.setdefault(key, copy_doc.id) + + for source in source_docs: + key = (source.observed, source.id) + if key in live_copies: + continue + plans.append( + _CopySpec( + observed=source.observed, + source_id=source.id, + content=source.content, + embedding=_embedding_as_list(source.embedding), + internal_metadata=dict(source.internal_metadata), + times_derived=source.times_derived, + source_ids=list(source.source_ids) + if source.source_ids is not None + else None, + session_name=session_name, + restore_document_id=soft_deleted_copies.get(key), + ) + ) + + if not plans: + return 0, set() + + # Phase 2 (no DB): fill missing embeddings. Source rows have NULL + # embeddings on external-store deployments (and soft-deleted copies may + # have lost their vectors) — re-embed via the embedding API only; no LLM. + missing = [spec for spec in plans if spec.embedding is None] + if missing: + logger.info( + "Scope backfill re-embedding %d documents with NULL embeddings for %s/%s/%s (embedding API only, no LLM)", + len(missing), + workspace_name, + scope_peer, + session_name, + ) + embeddings = await embedding_client.simple_batch_embed( + [spec.content for spec in missing] + ) + for spec, embedding in zip(missing, embeddings, strict=True): + spec.embedding = embedding + + # Phase 3 (DB): write the copies. + store_in_postgres = _store_embeddings_in_postgres() + touched_observed = {spec.observed for spec in plans} + new_rows: list[models.Document] = [] + async with tracked_db("scope_backfill.write") as db: + # scope_backfill and scope_removal carry different work-unit keys, so + # nothing orders them: a removal enqueued right after the add (or one + # that landed while phase 2 was embedding) can sweep the scope before + # these copies exist. Re-checking membership here, in the transaction + # that inserts, keeps a removed session from being copied back in. + if not await is_peer_in_session(db, workspace_name, session_name, scope_peer): + return None + + for observed in sorted(touched_observed): + await crud.get_or_create_collection( + db, workspace_name, observer=scope_peer, observed=observed + ) + + restore_ids: list[str] = [] + for spec in plans: + if spec.restore_document_id is not None: + restore_ids.append(spec.restore_document_id) + continue + row = models.Document( + workspace_name=workspace_name, + observer=scope_peer, + observed=spec.observed, + content=spec.content, + level="explicit", + times_derived=spec.times_derived, + internal_metadata={ + **spec.internal_metadata, + COPIED_FROM_KEY: spec.source_id, + }, + session_name=spec.session_name, + source_ids=spec.source_ids, + embedding=spec.embedding if store_in_postgres else None, + ) + row.sync_state = "pending" + new_rows.append(row) + + db.add_all(new_rows) + if restore_ids: + await db.execute( + update(models.Document) + .where(models.Document.id.in_(restore_ids)) + .values(deleted_at=None, sync_state="pending") + ) + await db.commit() + + # IDs are generated client-side and remain accessible post-commit + # (expire_on_commit=False), mirroring crud.document.create_documents. + for spec, row in zip( + [s for s in plans if s.restore_document_id is None], new_rows, strict=True + ): + spec.restore_document_id = row.id + + copied_ids = [ + spec.restore_document_id + for spec in plans + if spec.restore_document_id is not None + ] + + # Phase 4: sync to the external vector store (or mark synced in pgvector + # mode). Failures leave rows in sync_state='pending' for the reconciler. + await _sync_copies_to_vector_store(workspace_name, scope_peer, plans, copied_ids) + + return len(plans), touched_observed + + +async def _sync_copies_to_vector_store( + workspace_name: str, + scope_peer: str, + plans: list[_CopySpec], + copied_ids: list[str], +) -> None: + """Mirror the document-create sync path for the backfilled copies.""" + external_vector_store = get_external_vector_store() + + if external_vector_store is None: + # pgvector mode: embeddings live in the postgres column; nothing to sync. + async with tracked_db("scope_backfill.mark_synced") as db: + await db.execute( + update(models.Document) + .where(models.Document.id.in_(copied_ids)) + .values(sync_state="synced", last_sync_at=func.now(), sync_attempts=0) + ) + await db.commit() + return + + by_observed: dict[str, list[_CopySpec]] = {} + for spec in plans: + by_observed.setdefault(spec.observed, []).append(spec) + + synced_ids: list[str] = [] + failed_ids: list[str] = [] + for observed, specs in by_observed.items(): + namespace = external_vector_store.get_vector_namespace( + "document", workspace_name, scope_peer, observed + ) + records = [ + VectorRecord( + id=spec.restore_document_id, + embedding=spec.embedding, + metadata={ + "workspace_name": workspace_name, + "observer": scope_peer, + "observed": observed, + "session_name": spec.session_name, + "level": "explicit", + }, + ) + for spec in specs + if spec.restore_document_id is not None and spec.embedding is not None + ] + ids = [record.id for record in records] + try: + await external_vector_store.upsert_many(namespace, records) + synced_ids.extend(ids) + except Exception: + logger.exception( + "Failed to upsert backfilled vectors to %s; leaving docs pending for the reconciler", + namespace, + ) + failed_ids.extend(ids) + + async with tracked_db("scope_backfill.sync_state") as db: + if synced_ids: + await db.execute( + update(models.Document) + .where(models.Document.id.in_(synced_ids)) + .values(sync_state="synced", last_sync_at=func.now(), sync_attempts=0) + ) + if failed_ids: + await db.execute( + update(models.Document) + .where(models.Document.id.in_(failed_ids)) + .values( + sync_attempts=models.Document.sync_attempts + 1, + last_sync_at=func.now(), + ) + ) + await db.commit() + + +# --------------------------------------------------------------------------- +# Removal reconciliation (session removed from a scope) +# --------------------------------------------------------------------------- + + +async def process_scope_removal( + payload: ScopeRemovalPayload, workspace_name: str +) -> None: + """Process a ``scope_removal`` queue task. + + Soft-deletes (fail-closed) rather than hard-deletes: the reconciler's + soft-delete sweep handles vector cleanup and eventual hard deletion, and a + later re-add restores the same rows. + """ + scope_peer = payload.scope_peer + session_name = payload.session_name + + removed_by_observed: dict[str, list[str]] = {} + async with tracked_db("scope_removal") as db: + observed_result = await db.execute( + select(models.Collection.observed).where( + models.Collection.workspace_name == workspace_name, + models.Collection.observer == scope_peer, + ) + ) + for observed in [row[0] for row in observed_result.all()]: + explicit_stmt = ( + update(models.Document) + .where( + models.Document.workspace_name == workspace_name, + models.Document.observer == scope_peer, + models.Document.observed == observed, + models.Document.session_name == session_name, + models.Document.level == "explicit", + models.Document.deleted_at.is_(None), + ) + .values(deleted_at=func.now()) + .returning(models.Document.id) + ) + frontier = [row[0] for row in (await db.execute(explicit_stmt)).all()] + all_removed = list(frontier) + + # Fail-closed cascade: soft-delete derived documents whose support + # (source_ids) intersects anything removed, transitively — a + # deduction resting on removed evidence must leave with it, and so + # must an induction resting on that deduction. + while frontier: + derived_stmt = ( + update(models.Document) + .where( + models.Document.workspace_name == workspace_name, + models.Document.observer == scope_peer, + models.Document.observed == observed, + models.Document.level != "explicit", + models.Document.deleted_at.is_(None), + models.Document.source_ids.has_any(array(frontier)), + ) + .values(deleted_at=func.now()) + .returning(models.Document.id) + ) + frontier = [row[0] for row in (await db.execute(derived_stmt)).all()] + all_removed.extend(frontier) + + if all_removed: + removed_by_observed[observed] = all_removed + + # The session left the scope, so its backfill status entry is moot; + # clearing it keeps a later re-add starting from a fresh "pending". + await crud.clear_scope_backfill_status( + db, workspace_name, scope_peer, session_name + ) + await db.commit() + await crud.invalidate_scope_peer_cache(workspace_name, scope_peer) + + # Delete the vectors eagerly so recall can't surface removed memory while + # waiting for the reconciler sweep (which remains the backstop on failure). + external_vector_store = get_external_vector_store() + if external_vector_store is not None: + for observed, removed_ids in removed_by_observed.items(): + namespace = external_vector_store.get_vector_namespace( + "document", workspace_name, scope_peer, observed + ) + try: + await external_vector_store.delete_many(namespace, removed_ids) + except Exception: + logger.exception( + "Failed to delete removed vectors from %s; reconciler sweep will retry", + namespace, + ) + + # Rebuild what remains: the card must be regenerated from remaining + # evidence only (rebuild=True drops the stale card from the prompt), and a + # manual omni dream rebuilds the higher-order structure. + from src.deriver.enqueue import enqueue_dream + + for observed in sorted(removed_by_observed): + await enqueue_dream( + workspace_name, + observer=scope_peer, + observed=observed, + dream_type=DreamType.CARD_REFRESH, + rebuild=True, + trigger_reason="scope_removal", + ) + await enqueue_dream( + workspace_name, + observer=scope_peer, + observed=observed, + dream_type=DreamType.OMNI, + trigger_reason="scope_removal", + ) + + logger.info( + "Scope removal reconciliation complete for %s/%s/%s: %d documents soft-deleted across %d collections", + workspace_name, + scope_peer, + session_name, + sum(len(ids) for ids in removed_by_observed.values()), + len(removed_by_observed), + ) + + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- + + +async def _write_backfill_status( + workspace_name: str, + scope_peer: str, + session_name: str, + *, + state: ScopeBackfillState, + docs_copied: int | None = None, +) -> None: + """Best-effort status write in its own short-lived session.""" + try: + async with tracked_db("scope_backfill.status") as db: + await crud.update_scope_backfill_status( + db, + workspace_name, + scope_peer, + session_name, + state=state, + docs_copied=docs_copied, + ) + await db.commit() + await crud.invalidate_scope_peer_cache(workspace_name, scope_peer) + except Exception: + # Never mask the underlying failure (or fail a completed backfill) + # over a status bookkeeping write. + logger.exception( + "Failed to write scope backfill status %s for %s/%s/%s", + state, + workspace_name, + scope_peer, + session_name, + ) diff --git a/src/routers/scopes.py b/src/routers/scopes.py index 9d9087c0..de0d2865 100644 --- a/src/routers/scopes.py +++ b/src/routers/scopes.py @@ -8,9 +8,10 @@ that keeps the observer/observed mechanics hidden. All scopes routes require a workspace-level (or admin) key: scopes are an app-level admin surface, so peer- and session-scoped keys are rejected. -Note: scope membership only affects messages ingested *after* the membership -change. Conclusions already derived are neither backfilled on add nor -reconciled on removal. +Retroactive membership changes are handled asynchronously: adding a session +that already has messages enqueues a backfill-by-copy job, removing a session +enqueues a removal-reconciliation job. Track backfill progress via +``GET /scopes/{scope_id}/status``. """ import logging @@ -109,8 +110,9 @@ async def add_sessions_to_scope( is already a member is a no-op. List the resulting membership with `POST /scopes/{scope_id}/sessions/list`. - Note: membership applies only to messages ingested after this call; - conclusions already derived are not backfilled. + Note: any added session that already has messages triggers an asynchronous + backfill-by-copy of its existing documents into the scope; track progress + via ``GET /scopes/{scope_id}/status``. """ await crud.add_sessions_to_scope( db, @@ -135,8 +137,10 @@ async def remove_session_from_scope( """ Remove a Session from a Scope. - Note: conclusions already derived while the session was a member are left in - place. + Note: documents copied/derived while the session was a member are + reconciled asynchronously — the session's explicit copies are soft-deleted + from the scope, dependent derived documents follow (fail-closed), and the + scope's card is rebuilt from the remaining evidence. """ await crud.remove_session_from_scope( db, @@ -171,3 +175,24 @@ async def get_scope_sessions( workspace_name=workspace_id, scope_name=scope_id, reverse=reverse ), ) + + +@router.get( + "/{scope_id}/status", + response_model=schemas.ScopeStatus, + dependencies=[Depends(require_auth(workspace_name="workspace_id"))], +) +async def get_scope_status( + workspace_id: str = Path(...), + scope_id: str = Path(...), + db: AsyncSession = read_db, +): + """ + Get the backfill/reconciliation job status for a Scope. + + Returns a per-session map of the backfill job state (pending / completed / + failed) with the number of documents copied once complete. Empty when no + backfill has ever been enqueued for the scope. + """ + backfill_status = await crud.get_scope_backfill_status(db, workspace_id, scope_id) + return schemas.ScopeStatus(backfill_status=backfill_status) diff --git a/src/schemas/__init__.py b/src/schemas/__init__.py index 32e144b3..b4211b99 100644 --- a/src/schemas/__init__.py +++ b/src/schemas/__init__.py @@ -39,6 +39,7 @@ from src.schemas.api import ( Scope, ScopeCreate, ScopeSessionsAdd, + ScopeStatus, Session, SessionBase, SessionContext, @@ -140,6 +141,7 @@ __all__ = [ "Scope", "ScopeCreate", "ScopeSessionsAdd", + "ScopeStatus", "Session", "SessionBase", "SessionContext", diff --git a/src/schemas/api.py b/src/schemas/api.py index 411f2d14..55a6121a 100644 --- a/src/schemas/api.py +++ b/src/schemas/api.py @@ -428,9 +428,9 @@ class SessionCreate(SessionBase): max_length=100, description=( "Optional list of (unprefixed) scope names to add this session to. " - "Each scope is created if it does not exist yet. Membership applies " - "only to messages ingested after the session is added to the scope; " - "conclusions already derived are not backfilled." + "Each scope is created if it does not exist yet. If the session " + "already has messages, its existing documents are backfilled into " + "the scope asynchronously." ), ) @@ -591,6 +591,18 @@ class ScopeSessionsAdd(BaseModel): ) +class ScopeStatus(BaseModel): + """Per-session backfill/reconciliation job status for a scope. + + ``backfill_status`` maps each session that has had a backfill enqueued to + its current job state: ``{state, updated_at[, docs_copied]}`` where + ``state`` is ``pending``/``completed``/``failed`` and ``docs_copied`` is + present once a backfill completes. + """ + + backfill_status: dict[str, dict[str, Any]] = Field(default_factory=dict) + + # --------------------------------------------------------------------------- # Conclusion schemas # --------------------------------------------------------------------------- diff --git a/src/utils/queue_payload.py b/src/utils/queue_payload.py index 5d616b1e..59815ca5 100644 --- a/src/utils/queue_payload.py +++ b/src/utils/queue_payload.py @@ -73,6 +73,26 @@ class DreamPayload(BasePayload): rebuild: bool = False +class ScopeBackfillPayload(BasePayload): + """Payload for scope backfill tasks (session added to a scope, DEV-1999). + + workspace_name lives in the QueueItem's dedicated column, mirroring the + other task payloads. + """ + + task_type: Literal["scope_backfill"] = "scope_backfill" + scope_peer: str + session_name: str + + +class ScopeRemovalPayload(BasePayload): + """Payload for scope removal reconciliation tasks (session removed from a scope).""" + + task_type: Literal["scope_removal"] = "scope_removal" + scope_peer: str + session_name: str + + class DeletionPayload(BasePayload): """Payload for deletion tasks.""" @@ -124,6 +144,22 @@ def create_dream_payload( ).model_dump(mode="json", exclude_none=True) +def create_scope_task_payload( + task_type: Literal["scope_backfill", "scope_removal"], + *, + scope_peer: str, + session_name: str, +) -> dict[str, Any]: + """Create a scope backfill / removal payload.""" + if task_type == "scope_backfill": + return ScopeBackfillPayload( + scope_peer=scope_peer, session_name=session_name + ).model_dump(mode="json", exclude_none=True) + return ScopeRemovalPayload( + scope_peer=scope_peer, session_name=session_name + ).model_dump(mode="json", exclude_none=True) + + def create_deletion_payload( deletion_type: Literal["session", "observation", "workspace"], resource_id: str, diff --git a/src/utils/types.py b/src/utils/types.py index 33a2993d..24a13ae6 100644 --- a/src/utils/types.py +++ b/src/utils/types.py @@ -251,7 +251,14 @@ class GetOrCreateResult(Generic[T]): TaskType = Literal[ - "webhook", "summary", "representation", "dream", "deletion", "reconciler" + "webhook", + "summary", + "representation", + "dream", + "deletion", + "reconciler", + "scope_backfill", + "scope_removal", ] VectorSyncState = Literal["synced", "pending", "failed"] DocumentLevel = Literal["explicit", "deductive", "inductive", "contradiction"] diff --git a/src/utils/work_unit.py b/src/utils/work_unit.py index 6e0e25d4..2c05d58e 100644 --- a/src/utils/work_unit.py +++ b/src/utils/work_unit.py @@ -74,6 +74,15 @@ def construct_work_unit_key( raise ValueError("reconciler_type is required for reconciler tasks") return f"reconciler:{reconciler_type}" + if task_type in ("scope_backfill", "scope_removal"): + scope_peer = payload.get("scope_peer") + session_name = payload.get("session_name") + if not scope_peer or not session_name: + raise ValueError( + f"scope_peer and session_name are required for {task_type} tasks" + ) + return f"{task_type}:{workspace_name}:{scope_peer}:{session_name}" + raise ValueError(f"Invalid task type: {task_type}") @@ -183,4 +192,19 @@ def parse_work_unit_key(work_unit_key: str) -> ParsedWorkUnit: observed=None, ) + if task_type in ("scope_backfill", "scope_removal"): + # {task_type}:{workspace}:{scope_peer}:{session} + if len(parts) != 4: + raise ValueError( + f"Invalid work_unit_key format for task_type {task_type}: {work_unit_key}" + ) + return ParsedWorkUnit( + task_type=task_type, + workspace_name=parts[1], + session_name=parts[3], + # The scope peer is the observer of every collection the task touches. + observer=parts[2], + observed=None, + ) + raise ValueError(f"Invalid task type in work_unit_key: {task_type}") diff --git a/tests/conftest.py b/tests/conftest.py index b57e3786..f99859b3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -957,6 +957,7 @@ def mock_tracked_db(request: pytest.FixtureRequest): "src.dialectic.core.tracked_db", "src.dreamer.specialists.tracked_db", "src.dreamer.surprisal.tracked_db", + "src.deriver.scope_backfill.tracked_db", ] with ExitStack() as stack: for target in tracked_db_targets: diff --git a/tests/deriver/test_scope_backfill.py b/tests/deriver/test_scope_backfill.py new file mode 100644 index 00000000..06464c6f --- /dev/null +++ b/tests/deriver/test_scope_backfill.py @@ -0,0 +1,958 @@ +"""Tests for scope backfill-by-copy and removal reconciliation (DEV-1999). + +A scope is an observer peer (``scope.``). Adding a session with +pre-existing messages to a scope enqueues a ``scope_backfill`` task; the +handler (``src.deriver.scope_backfill``) copies each observed peer's +explicit-level documents from their global ``(P, P)`` collection into the +scope's ``(scope_peer, P)`` collection, stamping ``copied_from`` for +idempotency, then enqueues a manual omni dream. Removal enqueues +``scope_removal``, which soft-deletes the copies (cascading to dependent +derived documents) and enqueues a card_refresh (rebuild) + omni dream. + +These tests exercise the handlers directly (``process_scope_backfill`` / +``process_scope_removal``) against real Collection/Document/Peer rows, +mirroring the fixture style in tests/crud/test_document.py and +tests/dreamer/test_card_refresh.py: rows are created directly via +``db_session`` (never through the cache-backed ``crud.get_or_create_collection``, +which the ``mock_crud_collection_operations`` autouse fixture stubs out to an +unpersisted object for every other test). Fixture data must be *committed* +(not merely flushed) because the handlers run their DB work through +``tracked_db``, which in tests opens a separate session bound to the same +engine (see ``mock_tracked_db_context`` in conftest.py) — a different +connection that cannot see another session's uncommitted writes. +""" + +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from nanoid import generate as generate_nanoid +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models +from src.deriver.scope_backfill import ( + COPIED_FROM_KEY, + process_scope_backfill, + process_scope_removal, +) +from src.schemas import DreamType +from src.utils.queue_payload import ScopeBackfillPayload, ScopeRemovalPayload +from src.utils.scopes import is_scope_peer, scope_peer_name + +_EMBEDDING_DIM = 1536 + + +def _embedding(seed: float = 0.5) -> list[float]: + return [seed] * _EMBEDDING_DIM + + +async def _create_peer(db_session: AsyncSession, workspace_name: str) -> models.Peer: + peer = models.Peer(name=str(generate_nanoid()), workspace_name=workspace_name) + db_session.add(peer) + await db_session.commit() + return peer + + +async def _create_scope_peer( + db_session: AsyncSession, workspace_name: str, scope_name: str +) -> models.Peer: + peer = models.Peer( + name=scope_peer_name(scope_name), + workspace_name=workspace_name, + # The authoritative kind flag lives in internal_metadata, which is not + # user-writable; `configuration` carries only the observe_me knob. + internal_metadata={"kind": "scope"}, + configuration={"observe_me": False}, + ) + db_session.add(peer) + await db_session.commit() + return peer + + +async def _create_session( + db_session: AsyncSession, workspace_name: str +) -> models.Session: + session = models.Session(name=str(generate_nanoid()), workspace_name=workspace_name) + db_session.add(session) + await db_session.commit() + return session + + +async def _join_scope( + db_session: AsyncSession, workspace_name: str, session_name: str, scope_peer: str +) -> None: + """Record the scope peer's membership, as the scopes routes do. + + The backfill handler refuses to copy into a scope the session has left, so + tests driving the handler directly must stand the membership row up. + """ + db_session.add( + models.SessionPeer( + workspace_name=workspace_name, + session_name=session_name, + peer_name=scope_peer, + ) + ) + await db_session.commit() + + +async def _create_collection( + db_session: AsyncSession, workspace_name: str, observer: str, observed: str +) -> models.Collection: + collection = models.Collection( + workspace_name=workspace_name, observer=observer, observed=observed + ) + db_session.add(collection) + await db_session.commit() + return collection + + +async def _create_document( + db_session: AsyncSession, + workspace_name: str, + *, + observer: str, + observed: str, + session_name: str | None, + content: str = "some observation", + level: str = "explicit", + embedding: list[float] | None = None, + internal_metadata: dict[str, Any] | None = None, + source_ids: list[str] | None = None, +) -> models.Document: + doc = models.Document( + workspace_name=workspace_name, + observer=observer, + observed=observed, + content=content, + level=level, + session_name=session_name, + embedding=embedding if embedding is not None else _embedding(), + internal_metadata=internal_metadata or {}, + source_ids=source_ids, + ) + db_session.add(doc) + await db_session.commit() + return doc + + +async def _get_docs( + db_session: AsyncSession, + workspace_name: str, + *, + observer: str, + observed: str | None = None, + include_deleted: bool = True, +) -> list[models.Document]: + stmt = select(models.Document).where( + models.Document.workspace_name == workspace_name, + models.Document.observer == observer, + ) + if observed is not None: + stmt = stmt.where(models.Document.observed == observed) + if not include_deleted: + stmt = stmt.where(models.Document.deleted_at.is_(None)) + result = await db_session.execute(stmt) + return list(result.scalars().all()) + + +async def _dream_items( + db_session: AsyncSession, workspace_name: str +) -> list[models.QueueItem]: + result = await db_session.execute( + select(models.QueueItem).where( + models.QueueItem.workspace_name == workspace_name, + models.QueueItem.task_type == "dream", + ) + ) + return list(result.scalars().all()) + + +# --------------------------------------------------------------------------- +# 1. Backfill copies exactly the target session's explicit docs +# --------------------------------------------------------------------------- + + +async def test_backfill_copies_only_target_session_explicit_docs( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + + target_session = await _create_session(db_session, workspace_name) + other_session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, target_session.name, scope_peer.name) + + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + # Destination collection: crud.get_or_create_collection is stubbed to an + # unpersisted object by the autouse mock_crud_collection_operations + # fixture, so the scope's own collection must already exist for the + # copied Document rows' FK to resolve. + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + + # In-scope: the target session's explicit doc. + target_doc = await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=target_session.name, + content="target session explicit fact", + embedding=_embedding(0.7), + ) + # Out-of-scope: another session's explicit doc. + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=other_session.name, + content="other session explicit fact", + ) + # Out-of-scope: a derived (non-explicit) doc for the target session. + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=target_session.name, + content="deductive fact", + level="deductive", + ) + + await process_scope_backfill( + ScopeBackfillPayload( + scope_peer=scope_peer.name, session_name=target_session.name + ), + workspace_name, + ) + + copies = await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + assert len(copies) == 1 + copy = copies[0] + assert copy.content == "target session explicit fact" + assert copy.level == "explicit" + assert copy.session_name == target_session.name + assert copy.internal_metadata[COPIED_FROM_KEY] == target_doc.id + assert list(copy.embedding) == pytest.approx( # pyright: ignore[reportUnknownMemberType] + _embedding(0.7) + ) + assert copy.deleted_at is None + + +# --------------------------------------------------------------------------- +# 2. Idempotency +# --------------------------------------------------------------------------- + + +async def test_backfill_processed_twice_is_idempotent( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, session.name, scope_peer.name) + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=session.name, + ) + + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + + payload = ScopeBackfillPayload( + scope_peer=scope_peer.name, session_name=session.name + ) + await process_scope_backfill(payload, workspace_name) + await process_scope_backfill(payload, workspace_name) + + copies = await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + assert len(copies) == 1 + + +async def test_add_remove_readd_converges_on_one_live_copy( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, session.name, scope_peer.name) + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=session.name, + ) + + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + + backfill_payload = ScopeBackfillPayload( + scope_peer=scope_peer.name, session_name=session.name + ) + removal_payload = ScopeRemovalPayload( + scope_peer=scope_peer.name, session_name=session.name + ) + + # add + await process_scope_backfill(backfill_payload, workspace_name) + # remove + await process_scope_removal(removal_payload, workspace_name) + live = await _get_docs( + db_session, + workspace_name, + observer=scope_peer.name, + observed=sender.name, + include_deleted=False, + ) + assert live == [] + # re-add + await process_scope_backfill(backfill_payload, workspace_name) + + all_copies = await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + live_copies = [d for d in all_copies if d.deleted_at is None] + assert len(all_copies) == 1 # restored, not duplicated + assert len(live_copies) == 1 + + +async def test_backfill_skips_a_session_that_left_the_scope( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + """A removal that lands first must not be undone by a queued backfill. + + scope_backfill and scope_removal carry different work-unit keys, so nothing + orders them: add-then-remove can leave a backfill queued after removal has + already swept the scope. + """ + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, session.name, scope_peer.name) + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=session.name, + ) + + # The session leaves the scope before the queued backfill is drained. + await db_session.execute( + update(models.SessionPeer) + .where( + models.SessionPeer.workspace_name == workspace_name, + models.SessionPeer.session_name == session.name, + models.SessionPeer.peer_name == scope_peer.name, + ) + .values(left_at=func.now()) + ) + await db_session.commit() + + await process_scope_backfill( + ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name), + workspace_name, + ) + + assert ( + await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + == [] + ) + # No status entry either: removal cleared it, and a skipped backfill must + # not resurrect the session in the scope's status map. + # Names held as plain strings: expire_all() below would make reading them + # off the ORM instances trigger a lazy reload mid-assertion. + scope_peer_name_str, session_name = scope_peer.name, session.name + db_session.expire_all() + peer = await db_session.scalar( + select(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name == scope_peer_name_str) + ) + assert peer is not None + assert session_name not in peer.internal_metadata.get("backfill_status", {}) + + +# --------------------------------------------------------------------------- +# 3. Multi-peer session +# --------------------------------------------------------------------------- + + +async def test_backfill_multi_peer_session_copies_into_right_collections( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, peer_a = sample_data + workspace_name = test_workspace.name + peer_b = await _create_peer(db_session, workspace_name) + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, session.name, scope_peer.name) + + for peer in (peer_a, peer_b): + await _create_collection( + db_session, workspace_name, observer=peer.name, observed=peer.name + ) + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=peer.name + ) + await _create_document( + db_session, + workspace_name, + observer=peer.name, + observed=peer.name, + session_name=session.name, + content=f"fact about {peer.name}", + ) + + await process_scope_backfill( + ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name), + workspace_name, + ) + + copies_a = await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=peer_a.name + ) + copies_b = await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=peer_b.name + ) + assert len(copies_a) == 1 + assert copies_a[0].content == f"fact about {peer_a.name}" + assert len(copies_b) == 1 + assert copies_b[0].content == f"fact about {peer_b.name}" + + +# --------------------------------------------------------------------------- +# 4. Removal cascade +# --------------------------------------------------------------------------- + + +async def test_removal_cascades_to_dependent_derived_docs_only( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, session.name, scope_peer.name) + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=session.name, + ) + + await process_scope_backfill( + ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name), + workspace_name, + ) + [copy] = await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + + # A derived doc resting on the copy's evidence -> must be cascaded. + dependent = await _create_document( + db_session, + workspace_name, + observer=scope_peer.name, + observed=sender.name, + session_name=None, + content="deduction resting on removed evidence", + level="deductive", + source_ids=[copy.id], + ) + # An unrelated derived doc in the same collection -> must survive. + unrelated = await _create_document( + db_session, + workspace_name, + observer=scope_peer.name, + observed=sender.name, + session_name=None, + content="unrelated deduction", + level="deductive", + source_ids=["some-other-doc-id-not-removed"], + ) + + copy_id, dependent_id, unrelated_id = copy.id, dependent.id, unrelated.id + + await process_scope_removal( + ScopeRemovalPayload(scope_peer=scope_peer.name, session_name=session.name), + workspace_name, + ) + + # process_scope_removal runs on a separate tracked_db session (a + # different connection). Query raw columns rather than full ORM entities + # so this session's identity map (holding the pre-removal `copy` / + # `dependent` / `unrelated` instances) can't hand back stale, expired + # attributes. + result = await db_session.execute( + select(models.Document.id, models.Document.deleted_at).where( + models.Document.workspace_name == workspace_name, + models.Document.observer == scope_peer.name, + models.Document.observed == sender.name, + ) + ) + deleted_at_by_id = {row[0]: row[1] for row in result.all()} + assert deleted_at_by_id[copy_id] is not None + assert deleted_at_by_id[dependent_id] is not None + assert deleted_at_by_id[unrelated_id] is None + + +# --------------------------------------------------------------------------- +# 5. Dream enqueues +# --------------------------------------------------------------------------- + + +async def test_backfill_enqueues_manual_omni_dream( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, session.name, scope_peer.name) + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=session.name, + ) + + await process_scope_backfill( + ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name), + workspace_name, + ) + + dreams = await _dream_items(db_session, workspace_name) + assert len(dreams) == 1 + payload = dreams[0].payload + assert payload["dream_type"] == DreamType.OMNI.value + assert payload["observer"] == scope_peer.name + assert payload["observed"] == sender.name + assert payload["trigger_reason"] == "scope_backfill" + assert payload.get("rebuild", False) is False + + +async def test_removal_enqueues_card_refresh_rebuild_and_omni_dream( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + """Removal's own dream enqueues, isolated from backfill's. + + The scope's copy is created directly (as if an earlier backfill already + ran and its dream was drained by the deriver) rather than by calling + process_scope_backfill first: enqueue_dream dedupes on work_unit_key, so + a still-pending omni dream from an immediately-preceding backfill would + silently swallow removal's own omni enqueue and make this test couple to + that unrelated dedup behavior instead of testing removal in isolation. + """ + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + await _create_document( + db_session, + workspace_name, + observer=scope_peer.name, + observed=sender.name, + session_name=session.name, + internal_metadata={COPIED_FROM_KEY: "some-source-doc-id"}, + ) + + await process_scope_removal( + ScopeRemovalPayload(scope_peer=scope_peer.name, session_name=session.name), + workspace_name, + ) + + dreams = await _dream_items(db_session, workspace_name) + removal_dreams = [ + d for d in dreams if d.payload.get("trigger_reason") == "scope_removal" + ] + assert len(removal_dreams) == 2 + + by_type = {d.payload["dream_type"]: d.payload for d in removal_dreams} + assert DreamType.CARD_REFRESH.value in by_type + assert DreamType.OMNI.value in by_type + card_refresh_payload = by_type[DreamType.CARD_REFRESH.value] + assert card_refresh_payload["rebuild"] is True + assert card_refresh_payload["observer"] == scope_peer.name + assert card_refresh_payload["observed"] == sender.name + + +# --------------------------------------------------------------------------- +# 6. Status endpoint +# --------------------------------------------------------------------------- + + +async def test_status_reflects_pending_then_completed( + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + + response = client.post( + f"/v3/workspaces/{workspace_name}/scopes", json={"id": scope_name} + ) + assert response.status_code == 201 + scope_peer_full_name = scope_peer_name(scope_name) + + session_name = str(generate_nanoid()) + response = client.post( + f"/v3/workspaces/{workspace_name}/sessions", + json={"id": session_name, "peers": {sender.name: {}}}, + ) + assert response.status_code == 201 + + message = models.Message( + workspace_name=workspace_name, + session_name=session_name, + peer_name=sender.name, + content="hello from before the scope existed", + public_id=generate_nanoid(), + seq_in_session=1, + token_count=5, + ) + db_session.add(message) + await db_session.commit() + + # The message's explicit document (normally produced by the deriver) — + # created directly since the deriver isn't run in this test. + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=session_name, + ) + + response = client.post( + f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/sessions", + json={"session_ids": [session_name]}, + ) + assert response.status_code == 204, response.text + + # Destination collection: crud.get_or_create_collection is stubbed to an + # unpersisted object by the autouse fixture, so it must pre-exist. + await _create_collection( + db_session, workspace_name, observer=scope_peer_full_name, observed=sender.name + ) + + status_url = f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/status" + response = client.get(status_url) + assert response.status_code == 200, response.text + backfill_status = response.json()["backfill_status"] + assert backfill_status[session_name]["state"] == "pending" + + # Simulate the deriver picking up the enqueued task. + await process_scope_backfill( + ScopeBackfillPayload( + scope_peer=scope_peer_full_name, session_name=session_name + ), + workspace_name, + ) + + response = client.get(status_url) + assert response.status_code == 200, response.text + backfill_status = response.json()["backfill_status"] + assert backfill_status[session_name]["state"] == "completed" + assert backfill_status[session_name]["docs_copied"] == 1 + + +async def test_backfill_re_embeds_sources_with_null_embeddings( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + """Source rows carry no embedding on external-store deployments. + + Phase 2 re-embeds those (embedding API only) and pairs results back with + strict=True, so a mis-pairing would raise rather than silently mismatch. + """ + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + await _join_scope(db_session, workspace_name, session.name, scope_peer.name) + await _create_collection( + db_session, workspace_name, observer=sender.name, observed=sender.name + ) + await _create_collection( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + source = await _create_document( + db_session, + workspace_name, + observer=sender.name, + observed=sender.name, + session_name=session.name, + content="fact whose vector lives in the external store", + ) + source.embedding = None + await db_session.commit() + + await process_scope_backfill( + ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name), + workspace_name, + ) + + [copy] = await _get_docs( + db_session, workspace_name, observer=scope_peer.name, observed=sender.name + ) + assert copy.embedding is not None + assert len(copy.embedding) == _EMBEDDING_DIM + + +async def test_backfill_failure_records_failed_status( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + monkeypatch: pytest.MonkeyPatch, +): + test_workspace, _ = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name) + session = await _create_session(db_session, workspace_name) + + # Plain strings: the ORM instances are expired below (see the same guard in + # test_backfill_status_writes_preserve_the_scope_kind_flag). + scope_peer_name_str, session_name = scope_peer.name, session.name + + async def boom(*_args: Any, **_kwargs: Any) -> None: + raise RuntimeError("copy phase blew up") + + monkeypatch.setattr("src.deriver.scope_backfill._run_backfill", boom) + + with pytest.raises(RuntimeError): + await process_scope_backfill( + ScopeBackfillPayload( + scope_peer=scope_peer_name_str, session_name=session_name + ), + workspace_name, + ) + + db_session.expire_all() + peer = await db_session.scalar( + select(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name == scope_peer_name_str) + ) + assert peer is not None + assert peer.internal_metadata["backfill_status"][session_name]["state"] == "failed" + + +# --------------------------------------------------------------------------- +# 7. Route wiring: add-sessions enqueues backfill only when messages exist +# --------------------------------------------------------------------------- + + +async def test_add_sessions_enqueues_backfill_only_when_session_has_messages( + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + test_workspace, sender = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + assert ( + client.post( + f"/v3/workspaces/{workspace_name}/scopes", json={"id": scope_name} + ).status_code + == 201 + ) + + # Session with a pre-existing message. + session_with_messages = str(generate_nanoid()) + assert ( + client.post( + f"/v3/workspaces/{workspace_name}/sessions", + json={"id": session_with_messages, "peers": {sender.name: {}}}, + ).status_code + == 201 + ) + message = models.Message( + workspace_name=workspace_name, + session_name=session_with_messages, + peer_name=sender.name, + content="already said something", + public_id=generate_nanoid(), + seq_in_session=1, + token_count=5, + ) + db_session.add(message) + await db_session.commit() + + # Empty session, no messages. + empty_session = str(generate_nanoid()) + assert ( + client.post( + f"/v3/workspaces/{workspace_name}/sessions", + json={"id": empty_session}, + ).status_code + == 201 + ) + + response = client.post( + f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/sessions", + json={"session_ids": [session_with_messages, empty_session]}, + ) + assert response.status_code == 204, response.text + + result = await db_session.execute( + select(models.QueueItem).where( + models.QueueItem.workspace_name == workspace_name, + models.QueueItem.task_type == "scope_backfill", + ) + ) + backfill_items = list(result.scalars().all()) + assert len(backfill_items) == 1 + assert backfill_items[0].payload["session_name"] == session_with_messages + + +# --------------------------------------------------------------------------- +# The scope `kind` flag and the backfill status map share the scope peer's +# internal_metadata. Every write to that column must be a JSONB merge scoped to +# the backfill key; a wholesale assignment would drop the flag and silently turn +# the peer back into an ordinary one — invisible until some later read stopped +# recognising it as a scope. +# --------------------------------------------------------------------------- + + +async def test_backfill_status_writes_preserve_the_scope_kind_flag( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +): + """Status writes must not clobber the authoritative kind flag. + + Both live in internal_metadata, so this pins the one property that makes + them able to coexist. Covers the whole lifecycle, because a wholesale write + could be introduced at any single step: pending, completed, then cleared. + """ + test_workspace, _ = sample_data + workspace_name = test_workspace.name + scope_name = str(generate_nanoid()) + session_name = str(generate_nanoid()) + await _create_scope_peer(db_session, workspace_name, scope_name) + # Held as a plain string: the ORM instance is expired below on every check, + # so reading an attribute off it would trigger a reload mid-assertion. + backing_peer = scope_peer_name(scope_name) + + async def assert_still_a_scope(stage: str) -> dict[str, Any]: + db_session.expire_all() + refreshed = await db_session.scalar( + select(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name == backing_peer) + ) + assert refreshed is not None + assert is_scope_peer(refreshed.name, refreshed.internal_metadata), ( + f"the peer stopped being a scope after {stage}: " + f"internal_metadata={refreshed.internal_metadata!r}" + ) + # And the facade still resolves it, which is what actually breaks: + # get_scope_or_raise 404s on a peer that has lost the flag. + resolved = await crud.get_scope_or_raise(db_session, workspace_name, scope_name) + assert resolved.name == backing_peer + return refreshed.internal_metadata + + await assert_still_a_scope("creation") + + await crud.update_scope_backfill_status( + db_session, + workspace_name, + backing_peer, + session_name, + state="pending", + ) + await db_session.commit() + metadata = await assert_still_a_scope("a pending status write") + assert metadata["backfill_status"][session_name]["state"] == "pending" + + await crud.update_scope_backfill_status( + db_session, + workspace_name, + backing_peer, + session_name, + state="completed", + docs_copied=3, + ) + await db_session.commit() + metadata = await assert_still_a_scope("a completed status write") + assert metadata["backfill_status"][session_name]["docs_copied"] == 3 + + await crud.clear_scope_backfill_status( + db_session, workspace_name, backing_peer, session_name + ) + await db_session.commit() + metadata = await assert_still_a_scope("clearing the status") + assert session_name not in metadata.get("backfill_status", {})