mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(tinycortex): run the memory queue on the crate engine (W4) (#4781)
This commit is contained in:
Steven Enamakel
@@ -41,7 +41,7 @@ The crate exposes every seam the plan expects, plus more. Host implements these:
|
||||
|
||||
| Trait | Path | Host adapter (W1) |
|
||||
| --- | --- | --- |
|
||||
| `EmbeddingBackend` | `store/vectors/embedding.rs:32` | `tinycortex/embeddings.rs` over `openhuman::embeddings` |
|
||||
| `EmbeddingBackend` | `store/vectors/embedding.rs:32` | `tinycortex/embeddings.rs` over `openhuman::embeddings` *(amended: bridged to `tinyagents::harness::embeddings::EmbeddingModel` in W-EMB, plan §8.2)* |
|
||||
| `Embedder` (retrieval/summarise) | `score/embed.rs:28` | same seam file |
|
||||
| `ChatProvider` | `score/extract/llm.rs:70` | `tinycortex/chat.rs` over `memory::chat` / `inference` |
|
||||
| `Summariser` (bucket-seal, structured) | `tree/summarise.rs:70` | `tinycortex/chat.rs` |
|
||||
@@ -54,7 +54,9 @@ The crate exposes every seam the plan expects, plus more. Host implements these:
|
||||
| `EntityOccurrenceIndex` | `graph/types.rs:44` | `tinycortex/sinks.rs` |
|
||||
| `SelfIdentity` | `store/entity_index/store.rs:25` | host identity registry |
|
||||
| `GoalsGenerator` | `goals/reflect.rs:57` | `tinycortex/chat.rs` |
|
||||
| `SourceReader` | `sources/readers/mod.rs:1732` | local readers only; live sync stays host |
|
||||
| `SourceReader` | `sources/readers/mod.rs:1732` | local readers only *(amended: live-sync readers join the crate in W-SYNC, plan §8)* |
|
||||
| `SyncEventSink` *(new, W-SYNC)* | `sync/traits.rs` (planned) | `tinycortex/sync_sink.rs` → `MemorySyncStage` bus events |
|
||||
| `SkillDocSink` *(new, W-SYNC)* | `sync/traits.rs` (planned) | forwards to `MemoryClient::store_skill_sync` (host-retained unified tier) |
|
||||
| `ConversationEventBus` / `ChannelEventHandler` | `conversations/bus.rs:106/95` | `tinycortex/bus.rs` (translate to `DomainEvent`) |
|
||||
|
||||
Config seam: `MemoryConfig{workspace, embedding: EmbeddingConfig{dim,model,strict},
|
||||
@@ -165,3 +167,9 @@ retrieval: RetrievalConfig{default_profile}, sync_budget: SyncBudgetConfig}` +
|
||||
|
||||
**No hard gap blocks W1 (seam scaffolding) or W4 (queue, only drift D2).** W3 is gated by G1;
|
||||
W5 by G3/G6; W6/W7 by G2. These are recorded here and mirrored in the spec's deletion ledger.
|
||||
|
||||
**Amendment 2026-07-09 (plan §8):** the crate's sync data model (`SourceKind::Composio`,
|
||||
`SyncBudgetConfig`, the TOML source registry with `upsert_composio_source` /
|
||||
`memory_sync_defaults_for_toolkit`) — previously host-consumed only — becomes **engine-consumed by
|
||||
W-SYNC**: the crate's own sync engine reads the registry and enforces the budget. New seam traits
|
||||
`SyncEventSink` / `SkillDocSink` land with W-SYNC.1 (see roster above); no other new gaps.
|
||||
|
||||
@@ -25,9 +25,19 @@ touched an engine-mapping memory module after the port line, and classifies each
|
||||
| Host audit SHA | `7850cf363559bcbb7ba688cbc4fccdb6bd9ce754` (`main`, 2026-07-04) |
|
||||
| TinyCortex submodule | `vendor/tinycortex` → `tinyhumansai/tinycortex` |
|
||||
| TinyCortex audit SHA | `d1a8c7be2babc8fff7a72ed93861f459f3d6fa58` |
|
||||
| **TinyCortex pinned SHA (current)** | `a8e10f7dd8ebdb9b0905e1380fefcc6bf5a65207` — audit SHA **+ #59** (native-dep alignment, §0.4) **+ #63/#64** (D2/D1 drift ports) merged; this is the live gitlink |
|
||||
| TinyCortex crate version | `0.1.1` |
|
||||
| **Port line (derived)** | **after 2026-06-25, before 2026-06-28** (see below) |
|
||||
|
||||
> **Execution status (2026-07-10). Drift closure COMPLETE — all rows CLOSED.**
|
||||
> The submodule is pinned at `a8e10f7`. **D3 CLOSED** (folded into #59);
|
||||
> **D2 CLOSED** — merged as **tinycortex#63** (`is_host_io_error`, `e352435`);
|
||||
> **D1 CLOSED** — merged as **tinycortex#64** (rank-before-materialize, `a8e10f7`).
|
||||
> Host gitlink bumped `33dda94 → a8e10f7` (`chore(vendor): bump tinycortex …
|
||||
> (tinycortex#63, #64)`). With every drift row closed, **W4** (queue) and **W7**
|
||||
> (conversations) are now unblocked per the gate rule. D4 (memory_sync corpus,
|
||||
> W-SYNC) remains its own separate track.
|
||||
|
||||
### How the port line was located
|
||||
|
||||
The port commits in `vendor/tinycortex` are all dated **2026-06-29**, but that is the date the
|
||||
@@ -59,16 +69,16 @@ then per-commit file lists intersected with engine-mapping modules, then content
|
||||
|
||||
| # | Host commit | Module | Change | Crate state (verified) | Upstream target |
|
||||
| --- | --- | --- | --- | --- | --- |
|
||||
| D1 | `007a99b62` (06-30) `perf(memory_conversations): rank before cloning hits in cross-thread search` | `memory_conversations/inverted_index.rs` | Rank matches on cheap borrowed keys (`(doc_id:u32, matched:usize, created_at:&str)`), truncate to `limit`, **then** materialize the KB-sized `CrossThreadHit`. Order-equivalent to score ranking. | **ABSENT.** `vendor/tinycortex/src/memory/conversations/inverted_index.rs:286–301` builds the full `CrossThreadHit` (with `content.clone()`, `message_id.clone()`, `created_at.clone()`) for **every** matched doc, then `sort_by(score)` + `truncate`. Pre-fix clone-then-rank shape. | `conversations::inverted_index` — port the rank-before-materialize refactor + its `ranks_by_score_then_recency_before_truncating` test. |
|
||||
| D2 | `d7bee77e3` (06-30) `fix(memory-queue): classify host-FS I/O errors to stop the tree_jobs Sentry flood` | `memory_queue/worker.rs` | Adds `is_host_io_error(&anyhow::Error) -> bool` classifying **persistent** host-FS failures (EIO/ENOSPC/EROFS) distinct from transient SQLite busy/I-O, so the worker backs off and reports Sentry **once** instead of ~10k events/50min (Sentry CORE-RUST-19J). | **PARTIAL.** `vendor/tinycortex/src/memory/queue/worker.rs:89–107` has `is_sqlite_io_transient` (transient family) but **no** `is_host_io_error` (persistent host-FS family). | `queue::worker` — port the `is_host_io_error` predicate + its unit tests (EIO/ENOSPC/EROFS, context-layer, text fallback). **Only the predicate.** The Sentry-once emission and the `mark_storage_degraded` flag are host-owned (see D2-host below). |
|
||||
| D3 | `c43f79641` (07-03) (within TinyAgents migration) | `memory_store/vectors/store.rs` | `count()` reads `COUNT(*)` as `i64` and converts via `usize::try_from(...).context(...)` instead of `row.get::<usize>` directly — robustness against platform `usize`/`i64` mismatch. | **ABSENT.** `vendor/tinycortex/src/memory/store/vectors/store.rs:370–380` still does `let count: usize = ... row.get(0)` then `Ok(count)`. | `store::vectors::store` — small; port the `i64` + `try_from` guard. |
|
||||
| D1 | `007a99b62` (06-30) `perf(memory_conversations): rank before cloning hits in cross-thread search` | `memory_conversations/inverted_index.rs` | Rank matches on cheap borrowed keys (`(doc_id:u32, matched:usize, created_at:&str)`), truncate to `limit`, **then** materialize the KB-sized `CrossThreadHit`. Order-equivalent to score ranking. | ✅ **CLOSED.** Ported + merged as **tinycortex#64** (`a8e10f7`): the rank-before-materialize refactor + `ranks_by_score_then_recency_before_truncating` test. Was pre-fix clone-then-rank at the port line. | — (closed) |
|
||||
| D2 | `d7bee77e3` (06-30) `fix(memory-queue): classify host-FS I/O errors to stop the tree_jobs Sentry flood` | `memory_queue/worker.rs` | Adds `is_host_io_error(&anyhow::Error) -> bool` classifying **persistent** host-FS failures (EIO/ENOSPC/EROFS) distinct from transient SQLite busy/I-O, so the worker backs off and reports Sentry **once** instead of ~10k events/50min (Sentry CORE-RUST-19J). | ✅ **CLOSED.** Ported + merged as **tinycortex#63** (`e352435`): the `is_host_io_error` predicate + 2 unit tests (EIO/ENOSPC/EROFS × typed/context/text; negatives). **Predicate only** — the Sentry-once emission and `mark_storage_degraded` flag stay host-owned (see D2-host below). | — (closed) |
|
||||
| D3 | `c43f79641` (07-03) (within TinyAgents migration) | `memory_store/vectors/store.rs` | `count()` reads `COUNT(*)` as `i64` and converts via `usize::try_from(...).context(...)` instead of `row.get::<usize>` directly — robustness against platform `usize`/`i64` mismatch. | ✅ **CLOSED.** Present at `vendor/tinycortex/src/memory/store/vectors/store.rs:371–384` (`usize::try_from(count).context(...)`) — folded into #59's `usize`→`i64`/`try_from` rusqlite-0.40 sweep and merged. | — (closed) |
|
||||
|
||||
**Open drift rows: D1, D2 (predicate), D3.** These are the only three engine behavior changes since the
|
||||
port line. All three are small-to-moderate and independent.
|
||||
**Drift rows: D1, D2 (predicate), D3** — the only three engine behavior changes since the port line,
|
||||
all small and independent, **now all CLOSED** (gitlink `a8e10f7`).
|
||||
|
||||
- D1 gates **W7** (long tail — conversations).
|
||||
- D2 gates **W4** (queue).
|
||||
- D3 gates **W3** (store + chunks).
|
||||
- D1 gates **W7** (long tail — conversations). ✅ **CLOSED** — merged as tinycortex#64.
|
||||
- D2 gates **W4** (queue). ✅ **CLOSED** — merged as tinycortex#63.
|
||||
- D3 gates **W3** (store + chunks). ✅ **CLOSED** — already in the crate via #59.
|
||||
|
||||
### HOST-OWNED — same commits, layers that stay in OpenHuman (no upstream)
|
||||
|
||||
@@ -102,10 +112,10 @@ not as drift. See `tinycortex-api-gap-audit.md`.
|
||||
|
||||
| Engine module | Maps to crate | Open drift | Gates workstream | Status |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| `memory_store` (chunks, content, vectors, kv, entity_index, safety) | `store/`, `chunks/` | **D3** (vectors count guard). `wiki_git`/`obsidian*` host-retained (not drift). | W3 | **OPEN** (D3) |
|
||||
| `memory_store` (chunks, content, vectors, kv, entity_index, safety) | `store/`, `chunks/` | **D3** (vectors count guard) ✅ closed via #59. `wiki_git`/`obsidian*` host-retained (not drift). | W3 | **CLEAR** (D3 closed) |
|
||||
| `memory_tree` (tree, retrieval, score, summarise) | `tree/`, `retrieval/`, `score/` | none (health/rpc/embed-compute are host-owned) | W5 | **CLEAR** |
|
||||
| `memory_queue` | `queue/` | **D2** (predicate) | W4 | **OPEN** (D2) |
|
||||
| `memory_conversations` | `conversations/` | **D1** (rank-before-clone) | W7 | **OPEN** (D1) |
|
||||
| `memory_queue` | `queue/` | **D2** (predicate) — ✅ merged tinycortex#63 | W4 | **CLEAR** (D2 closed) |
|
||||
| `memory_conversations` | `conversations/` | **D1** (rank-before-clone) — ✅ merged tinycortex#64 | W7 | **CLEAR** (D1 closed) |
|
||||
| `memory_diff` | `diff/` | none (git-ledger captured) | W7 | **CLEAR** |
|
||||
| `memory_entities` | `entities/` | none | W7 | **CLEAR** |
|
||||
| `memory_graph` | `graph/` | none | W7 | **CLEAR** |
|
||||
@@ -115,10 +125,34 @@ not as drift. See `tinycortex-api-gap-audit.md`.
|
||||
| `memory_tools` (engine part) | `tool_memory/` | none | W7 | **CLEAR** |
|
||||
| `memory_search` (`vector`, `scoring` engine parts; `tools` are host) | `retrieval/`, `score/` | none (churn only) | W5 | **CLEAR** (classify tools vs engine in W5) |
|
||||
|
||||
**Summary:** 3 open drift rows (D1, D2, D3), each small and independent, each a single-module
|
||||
tinycortex PR. Nothing else drifted. `memory_search` is a mixed module not in the plan's move table —
|
||||
**Summary:** 3 drift rows total, **all CLOSED** — **D3** via #59, **D2** via tinycortex#63,
|
||||
**D1** via tinycortex#64; gitlink now `a8e10f7`. No engine drift remains open; **W4 and W7 are
|
||||
unblocked**. Nothing else drifted. `memory_search` is a mixed module not in the plan's move table —
|
||||
its `tools/` stay host (agent tools), its `vector`/`scoring` are engine (W5) — flagged for the gap audit.
|
||||
|
||||
## D4 — memory_sync corpus (2026-07-09 reclassification, W-SYNC)
|
||||
|
||||
The plan's §8 amendment moves the sync engine into the crate, so the blanket **HOST-OWNED
|
||||
"live sync"** classification above is superseded for engine-mapping sync code. Unlike D1–D3 there
|
||||
is **no existing crate port to diff against** — the entire `memory_sync/` engine (plus the
|
||||
`memory_sources` dispatcher/reconcile parts) is port scope. The port baseline for W-SYNC is taken
|
||||
**fresh from host `main` at W-SYNC.1 branch time**, so ordinary drift cannot accumulate; this entry
|
||||
exists to (a) record the reclassification and (b) pin the post-port-line commits that already
|
||||
touched the corpus, so the W-SYNC.1 port provably includes them:
|
||||
|
||||
| # | Host commit | Files in corpus | Note |
|
||||
| --- | --- | --- | --- |
|
||||
| D4.1 | `c43f79641` (07-03) | `composio/providers/{sync_state,traits}.rs` | TinyAgents-cutover import churn |
|
||||
| D4.2 | `27b00b539` (07-05) | `sources/rebuild.rs` | test-parity cleanup (1 line) |
|
||||
| D4.3 | `653e6e143` (07-06) | `memory_sources/sync.rs` (+312) | self-heal drifted content-sha tokens + prune vanished folder items |
|
||||
| D4.4 | `e456b7799` (07-07) | `memory_sources/{rpc,sync}.rs`, `canonicalize/email.rs`, `composio/providers/{gmail/post_process,notion/source,orchestrator}.rs`, `sources/github.rs` | orchestration-fixes wave |
|
||||
|
||||
**Gate:** W-SYNC.1 (crate scaffolding PR) requires this enumeration current as of its branch point
|
||||
(re-run the scan: `git log --since=2026-06-25 --oneline -- src/openhuman/memory_sync
|
||||
src/openhuman/memory_sources`); the W-SYNC.3 host flip requires D4 **CLOSED** — every listed commit
|
||||
(and any accrued since) verifiably contained in the crate port or explicitly waived. Host-retained
|
||||
parts (schedulers, bus subscribers, RPC wrappers, keychain/OAuth) remain HOST-OWNED as before.
|
||||
|
||||
## Closing the ledger (procedure)
|
||||
|
||||
For each open row:
|
||||
|
||||
@@ -1,9 +1,16 @@
|
||||
# TinyCortex Memory Migration — Plan & Audit
|
||||
|
||||
**Status:** draft plan — no code changes yet.
|
||||
**Status:** W1–W2 landed; W3 partial (chunks sub-store flipped). Amended 2026-07-09 (see §8) to a **sync-inclusive** scope.
|
||||
**Anchor precedent:** the TinyAgents harness migration (#4249 / #4399 / #4473 and follow-ups).
|
||||
**Target:** migrate large portions of the OpenHuman memory subsystem onto the `tinycortex` crate, vendored as a git submodule at **`vendor/tinycortex`** (`https://github.com/tinyhumansai/tinycortex`).
|
||||
|
||||
> **Amendment 2026-07-09 (§8 — W-SYNC + W-EMB).** The original seam contract kept live sync
|
||||
> (`memory_sync/`) host-side and TinyCortex strictly network-free. That boundary is **revised**:
|
||||
> TinyCortex becomes all-encompassing of the sync modules (Composio toolkit sync included) behind an
|
||||
> optional `sync` cargo feature, and embeddings are inherited from `tinyagents::harness::embeddings`.
|
||||
> Sections §1/§3 below are the original contract, kept for history; where they conflict with §8,
|
||||
> **§8 wins**.
|
||||
|
||||
---
|
||||
|
||||
## 0. Ground truth (as audited)
|
||||
@@ -63,7 +70,7 @@ TinyCortex lives at **`vendor/tinycortex`** as a git submodule. Any change to en
|
||||
|
||||
- **All RPC surfaces**: `memory/ops/`, `memory/schemas/`, `memory/schema/`, `memory/read_rpc/`, `rpc_models.rs` (controller framework types `ControllerSchema`/`RpcOutcome` are host-only). JSON-RPC method names and payload shapes must not change.
|
||||
- **Agent tools**: `memory/tools/` and `memory/query/` (`Tool`/`ToolResult` impls, `SecurityPolicy` gating) — they become thin wrappers over crate retrieval primitives.
|
||||
- **Live sync**: all of `memory_sync/` (Composio/MCP/OAuth/pollers), `memory/sync.rs` lifecycle + event-bus stage events.
|
||||
- **Live sync**: all of `memory_sync/` (Composio/MCP/OAuth/pollers), `memory/sync.rs` lifecycle + event-bus stage events. *(Superseded by §8: the sync engine moves to the crate; only schedulers, credentials/OAuth, bus bridges, and RPC wrappers stay host.)*
|
||||
- **Process glue**: `memory/global.rs` singleton + background queue worker; `memory/source_scope.rs` tokio task-locals; `memory/chat.rs` (LLM adapter over `openhuman::inference`); embeddings provider wiring.
|
||||
- **Policy/UX**: `preferences.rs`, `remember.rs`, `tree_policy.rs`, `util/redact.rs`, config mapping (`Config` → `tinycortex::MemoryConfig`).
|
||||
- **Namespace document/graph store** (until/unless deliberately upstreamed — TinyCortex explicitly excludes it today).
|
||||
@@ -116,7 +123,7 @@ Per the tinyagents rules: **adapter first → prove parity → flip ownership
|
||||
|
||||
**W6 — Ingest.** `memory/ingest_pipeline.rs` + `memory/ingestion/` re-pointed to `tinycortex::ingest` + `score::extract` (LLM extraction via the seam's `ChatProvider`). The namespace document/graph store path stays host-side unless deliberately upstreamed (explicit decision in this workstream). `ingest_chat`/`ingest_document_with_scope` keep their signatures — 11 call sites (learning, agent harness, archivist) unchanged.
|
||||
|
||||
**W7 — Long tail.** `memory_diff`→`diff`, `memory_entities`→`entities`, `memory_graph`→`graph`, `memory_goals`→`goals`, `memory_archivist`→`archivist`, `memory_sources` registry/local readers→`sources`, tool-memory engine→`tool_memory`, conversation storage→`conversations`. Each is small and independent; each deletes its legacy module on flip. `memory_sync` explicitly **does not move** — it keeps writing through the crate's ingest/source contracts.
|
||||
**W7 — Long tail.** `memory_diff`→`diff`, `memory_entities`→`entities`, `memory_graph`→`graph`, `memory_goals`→`goals`, `memory_archivist`→`archivist`, `memory_sources` registry/local readers→`sources`, tool-memory engine→`tool_memory`, conversation storage→`conversations`. Each is small and independent; each deletes its legacy module on flip. `memory_sync` engine parts **move via W-SYNC (§8)** — the original "does not move" rule is superseded.
|
||||
|
||||
**W8 — Test port + parity sweep + deletion-ledger close-out.** See §5. Ends with: deletion ledger fully executed, `gitbooks/developing/architecture.md` + a new `architecture/memory.md` seam doc written (the durable post-plan documentation, as `agent-harness.md` was for tinyagents), `AGENTS.md`/`CLAUDE.md` module tables updated, and the spec doc archived.
|
||||
|
||||
@@ -173,3 +180,86 @@ Ordering rule for this migration: **within each workstream, the implementation (
|
||||
- All engine logic served by `tinycortex` at a tagged, crates.io-published version, submodule pinned in lockstep.
|
||||
- JSON-RPC method names/payloads unchanged; existing user workspaces open and recall identically (golden-workspace parity green).
|
||||
- Full suites green on both CI lanes; gitbooks/AGENTS.md updated; spec + ledgers archived.
|
||||
- **(Amended)** `memory_sync/` engine deleted from the host per §8 (W-SYNC); host retains only
|
||||
schedulers, credentials/OAuth, event-bus bridges, and RPC wrappers. `src/openhuman/embeddings/`
|
||||
provider impls deleted per §8 (W-EMB); embeddings served by `tinyagents::harness::embeddings`.
|
||||
|
||||
---
|
||||
|
||||
## 8. Amendment (2026-07-09) — sync-inclusive scope + embeddings inheritance
|
||||
|
||||
Decisions (confirmed with the maintainer): TinyCortex becomes **all-encompassing of the sync
|
||||
modules**; "Composio-related skills syncing" = the per-toolkit memory-sync providers
|
||||
(`memory_sync/composio/providers/{gmail,slack,github,notion,linear,clickup}`); **full network in
|
||||
crate** behind a cargo feature; Composio API key via env var + config field (host still resolves
|
||||
from the `composio-direct` keychain slot in production); a **mocked + live test pair** gates the
|
||||
Composio sync port; embeddings **fully inherited from TinyAgents**.
|
||||
|
||||
### 8.1 Revised seam contract — W-SYNC
|
||||
|
||||
**Moves into TinyCortex** (new cargo feature `sync`, off by default so the default build stays a
|
||||
pure, network-free library):
|
||||
|
||||
| Host source | Crate destination |
|
||||
| --- | --- |
|
||||
| `memory_sync/traits.rs` (`SyncPipeline`/`SyncOutcome`/`SyncPipelineKind`) | `src/memory/sync/traits.rs` (init/tick take `&MemoryConfig` + a `SyncContext`) |
|
||||
| `memory_sync/composio/providers/*` (all 6 toolkits + registry, orchestrator, sync_state, catalogs, user_scopes) | `src/memory/sync/composio/providers/*` |
|
||||
| New Composio HTTP client (modeled on `src/openhuman/composio/client.rs`, minus keychain) | `src/memory/sync/composio/client.rs` — **direct** (BYO key → backend.composio.dev) and **proxied** (base_url + bearer; OpenHuman-backend default) modes |
|
||||
| `memory_sync/canonicalize/` | merged into the crate's existing `ingest/canonicalize` |
|
||||
| `memory_sync/workspace/` scan logic (not the timers) | `src/memory/sync/workspace.rs` |
|
||||
| `memory_sync/sources/{audit,rebuild}.rs` | `src/memory/sync/{audit,rebuild}.rs` |
|
||||
| `memory_sync/sync_status/` SQL | `src/memory/sync/status.rs` (chunk schema lives in the crate) |
|
||||
| `memory_sources/sync.rs::sync_source` dispatcher + `reconcile.rs::ensure_composio_sources` | `src/memory/sync/dispatch.rs` |
|
||||
|
||||
**Stays host-side:** scheduler loops (`composio/periodic.rs`, `workspace/periodic.rs`, file
|
||||
watcher — crate exposes `sync::tick_all`/`tick_pipeline(id)`, host owns tokio timers, same pattern
|
||||
as `queue::run_once`); credentials/OAuth (keychain slot `composio-direct`, RPC
|
||||
`composio.{get_mode,set_api_key,clear_api_key}`, connection lifecycle); event bus (new
|
||||
**`SyncEventSink`** trait in the crate, host adapter `src/openhuman/tinycortex/sync_sink.rs`
|
||||
translates to `MemorySyncStage` bus events); RPC wrappers (`memory/ops/sync.rs`,
|
||||
`memory/schemas/sync.rs`, `memory_sources/rpc.rs` — JSON-RPC names/payloads unchanged); the
|
||||
UnifiedMemory writeback path via a new **`SkillDocSink`** trait (providers call
|
||||
`MemoryClient::store_skill_sync`, which writes the host-retained namespace-document tier — the
|
||||
crate must not depend on that tier); MCP transport (MCP pipeline stays host through W-SYNC.3).
|
||||
|
||||
**Config/key:** crate gains `SyncConfig { budget, composio: Option<ComposioSyncConfig> }` with
|
||||
`ComposioSyncConfig { mode: Direct|Proxied, base_url, api_key: Option<SecretString>, bearer_token,
|
||||
entity_id }`; `api_key` falls back to the `COMPOSIO_API_KEY` env var (tests only — the app resolves
|
||||
from keychain and injects). `SecretString` redacts `Debug`/`Display` and skips serialization.
|
||||
|
||||
**Tests (gate):** (a) `vendor/tinycortex/tests/composio_sync_mock.rs` — wiremock-backed gmail
|
||||
pipeline test, always-on in crate CI (pagination, cursor advance, taint = `external_sync`,
|
||||
idempotent second tick, `SkillDocSink` capture, 401 without key leakage); (b)
|
||||
`vendor/tinycortex/tests/composio_sync_live.rs` — `#[ignore]`, runs when `COMPOSIO_API_KEY` is set,
|
||||
direct-mode gmail sync with structural assertions, budget-capped (~5 records).
|
||||
|
||||
**Ordering:** W-SYNC slots **after W6, before W7** (providers write through crate ingest; queue
|
||||
jobs through crate queue). Phases: **W-SYNC.1** scaffolding (feature + config + client + traits +
|
||||
gmail provider) → **W-SYNC.2** remaining providers + workspace/audit/rebuild/status → **W-SYNC.3**
|
||||
host flip (delete `memory_sync` bodies; wire adapters/schedulers/RPC wrappers) → **W-SYNC.4**
|
||||
cleanup (dedupe the agent-tools Composio client, remove `composio/{bus,periodic,providers}.rs`
|
||||
shims, decide MCP).
|
||||
|
||||
**Drift:** the HOST-OWNED "live sync" classification in the drift ledger is superseded — see the
|
||||
ledger's **D4** entry.
|
||||
|
||||
### 8.2 Embeddings inherited from TinyAgents — W-EMB
|
||||
|
||||
`tinyagents::harness::embeddings` becomes the canonical embeddings module:
|
||||
|
||||
- **W-EMB.1 (tinyagents PR):** extend `EmbeddingModel` with `name()`/`model_id()`/`signature()`
|
||||
(default **byte-identical** to the host format `provider={name};model={model_id};dims={dims}` —
|
||||
parity P10 / #1574); port host providers (`ollama`, `cohere`, `voyage`, `cloud`, `noop`) +
|
||||
`rate_limit.rs`/`retry_after.rs` into `harness::embeddings::providers/*`; merge host `openai.rs`
|
||||
into the existing `OpenAiEmbeddingModel`.
|
||||
- **W-EMB.2 (tinycortex PR):** add a `tinyagents` dependency; bridge/replace `EmbeddingBackend`
|
||||
with `tinyagents::harness::embeddings::EmbeddingModel` (re-export or blanket impl).
|
||||
- **W-EMB.3 (host PR):** dual submodule bump; `src/openhuman/tinycortex/embeddings.rs` constructs
|
||||
tinyagents providers from `Config`; delete `src/openhuman/embeddings/` provider impls; keep
|
||||
`factory.rs` (thin), `rpc.rs`, `schemas.rs`, catalog + config wiring host-side; signature-parity
|
||||
seam tests in the same PR.
|
||||
|
||||
W-EMB is independent of W-SYNC and may run in parallel; land W-EMB.2 **before** the W-SYNC.3 flip
|
||||
so `SyncContext` is defined against the tinyagents trait once. New risk: `tinycortex → tinyagents`
|
||||
crate coupling — version bumps move in lockstep in both `[patch.crates-io]` blocks (root +
|
||||
`app/src-tauri`).
|
||||
|
||||
@@ -1,7 +1,11 @@
|
||||
# TinyCortex Memory Migration — Spec (Phase 0.5 / 0.6)
|
||||
|
||||
**Status:** Phase 0 baseline. Anchors the migration to exact reviewed SHAs and ties together the
|
||||
drift, gap, and parity ledgers. Modeled on `docs/tinyagents-migration-spec.md` + its deletion ledger.
|
||||
**Status:** Phase 0 baseline **+ execution underway** — #59 (native-dep alignment) merged and the dep
|
||||
is active (§0.4); W1 seam + W2 type re-export + W3-chunks partial landed; **drift closure COMPLETE**
|
||||
(**D3** via #59, **D2** via tinycortex#63, **D1** via tinycortex#64; gitlink `a8e10f7`) — so **W4
|
||||
(queue) and W7 (conversations) are unblocked**. Anchors the migration to exact reviewed SHAs and ties
|
||||
together the drift, gap, and parity ledgers. Modeled on `docs/tinyagents-migration-spec.md` + its
|
||||
deletion ledger.
|
||||
|
||||
**Companion plan:** [`tinycortex-memory-migration-plan.md`](tinycortex-memory-migration-plan.md)
|
||||
**Ledgers:** [`tinycortex-drift-ledger.md`](tinycortex-drift-ledger.md) ·
|
||||
@@ -14,7 +18,8 @@ drift, gap, and parity ledgers. Modeled on `docs/tinyagents-migration-spec.md` +
|
||||
| --- | --- | --- |
|
||||
| `tinyhumansai/openhuman` | `7850cf363559bcbb7ba688cbc4fccdb6bd9ce754` | host audit base (`main`, 2026-07-04) |
|
||||
| `tinyhumansai/tinycortex` | `d1a8c7be2babc8fff7a72ed93861f459f3d6fa58` | crate audit base (v0.1.1) |
|
||||
| `tinyhumansai/tinycortex` | `35f9518` (branch `chore/align-native-deps-for-openhuman`) | **first required upstream PR** — native-dep alignment (§0.4); host may not activate the dep until this merges |
|
||||
| `tinyhumansai/tinycortex` | `33dda943053e61ef585fc39647cf1854344b6323` | audit base **+ #59** (native-dep alignment, §0.4) merged |
|
||||
| `tinyhumansai/tinycortex` | `a8e10f7dd8ebdb9b0905e1380fefcc6bf5a65207` | **current pinned gitlink** — **+ #63/#64** (D2/D1 drift closure) merged; all drift rows CLOSED |
|
||||
|
||||
Port line (derived by content, §0.1): **after 2026-06-25, before 2026-06-28** for engine features.
|
||||
|
||||
@@ -64,18 +69,19 @@ ledger. Re-export covers the data types (`MemoryEntry`, `MemoryCategory`, `Memor
|
||||
| Crate edition | 2021 (matches host) ✅ |
|
||||
| `[patch.crates-io] tinycortex = { path = "vendor/tinycortex" }` | pre-staged in **both** worlds (root + `app/src-tauri`) ✅ |
|
||||
| CI submodule checkout | recursive on all build/test lanes; covers `vendor/tinycortex` ✅ (verify release lanes in W1, as tinyagents needed) |
|
||||
| **rusqlite alignment** | ⚠️ **blocker resolved upstream.** Crate pinned `0.32` (bundled), host pins `=0.40.0` (bundled). Two `links = "sqlite3"` = hard Cargo error. Fixed in the crate PR (bump to `0.40` + `usize`→`i64`/`try_from`). |
|
||||
| **git2 alignment** | ⚠️ **blocker resolved upstream.** Crate pinned `0.19`, host `0.21` (vendored-libgit2). Two `links = "git2"`. Fixed in the crate PR (bump to `0.21` + API deltas: `Tag::message`, `StringArray::Iter`, `Buf::as_str`). |
|
||||
| **rusqlite alignment** | ✅ **resolved & merged (#59).** Crate was pinned `0.32` (bundled), host pins `=0.40.0` (bundled). Two `links = "sqlite3"` = hard Cargo error. Fixed in #59 (bump to `0.40` + `usize`→`i64`/`try_from` — the same sweep that closed drift **D3**). |
|
||||
| **git2 alignment** | ✅ **resolved & merged (#59).** Crate was pinned `0.19`, host `0.21` (vendored-libgit2). Two `links = "git2"`. Fixed in #59 (bump to `0.21` + API deltas: `Tag::message`, `StringArray::Iter`, `Buf::as_str`). |
|
||||
| Crate compiles with aligned deps | ✅ `cargo check --all-targets` clean; 38 diff/checkpoint tests pass. |
|
||||
| **Host root world compiles with dep active** | ✅ `cargo check --manifest-path Cargo.toml --lib` **exit 0** with `tinycortex = "0.1"` active + submodule at `35f9518`. **No `multiple packages link to native library` error** — one bundled SQLite + one libgit2 confirmed. (Verified locally; the Cargo.toml/lock activation is reverted from the Phase-0 docs branch and re-lands in W1 post-upstream-merge.) |
|
||||
| **Host root world compiles with dep active** | ✅ `cargo check --manifest-path Cargo.toml --lib` **exit 0** with `tinycortex = "0.1"` active (`Cargo.toml:82`) + submodule at `33dda94`. **No `multiple packages link to native library` error** — one bundled SQLite + one libgit2 confirmed. **Now landed** (post-#59-merge): the `[dependencies]` line and gitlink are committed on the working branch, not reverted. Re-verified 2026-07-09. |
|
||||
| Host `app/src-tauri` world | to verify in W1 (separate Cargo world / lockfile). |
|
||||
| `GGML_NATIVE=OFF` macOS ARM | to verify on a macOS runner in W1 (no macOS host here). |
|
||||
|
||||
**Activation is deferred to W1.** Per the submodule rule, the host may only bump the gitlink to a
|
||||
**merged** upstream SHA. The native-dep alignment is committed on the crate branch and ready to PR;
|
||||
once it merges and is published, W1 lands: (a) `chore(vendor): bump tinycortex`, (b)
|
||||
`[dependencies] tinycortex = "0.1"` in both worlds. The `[dependencies]` line and the compile
|
||||
verification below were validated **locally** against the branch to prove the baseline is sound.
|
||||
**Activation landed (post-#59).** The native-dep alignment merged upstream as **#59** and the host
|
||||
gitlink was bumped to `33dda94`, so — per the submodule rule (bump only to a **merged** SHA) — W1's
|
||||
activation is now in place: `[dependencies] tinycortex = "0.1"` is active in the root world
|
||||
(`Cargo.toml:82`), the seam (`src/openhuman/tinycortex/`) is wired (`src/openhuman/mod.rs:130`), and
|
||||
`cargo check --lib` is **exit 0**. Remaining §0.4 follow-ups: the `app/src-tauri` world and the
|
||||
`GGML_NATIVE=OFF` macOS-runner check still verify in W1 (see rows above).
|
||||
|
||||
---
|
||||
|
||||
@@ -96,8 +102,16 @@ verification below were validated **locally** against the branch to prove the ba
|
||||
|
||||
- **RPC surfaces:** `memory/{ops,schemas,schema,read_rpc}`, `rpc_models.rs`. Method names/payloads unchanged.
|
||||
- **Agent tools:** `memory/tools/`, `memory/query/`, `memory_search/tools/`, `memory_tools`(tool surface) — thin wrappers over crate retrieval + `SecurityPolicy` gating.
|
||||
- **Live sync:** all of `memory_sync/`, `memory/sync.rs` lifecycle + bus stage events.
|
||||
- **Process glue:** `memory/global.rs` singleton + queue worker; `memory/source_scope.rs` task-locals; `memory/chat.rs`; embeddings provider wiring.
|
||||
- **Live sync (amended 2026-07-09, plan §8 / W-SYNC):** the sync **engine** (pipelines, per-toolkit
|
||||
Composio providers + HTTP client, canonicalize, sync_state, audit/rebuild, sync_status query,
|
||||
dispatcher) **moves to the crate** behind an optional `sync` cargo feature. The crate's
|
||||
"never makes a network call" invariant becomes **feature-scoped**: the default build stays
|
||||
network-free; the `sync` feature adds an HTTP Composio client whose credentials are injected by
|
||||
the host. Host retains: scheduler loops (tick-driven crate, like `queue::run_once`),
|
||||
credentials/OAuth (keychain `composio-direct`), event-bus bridges via a new **`SyncEventSink`**
|
||||
seam trait, RPC wrappers (`memory/{ops,schemas}/sync.rs`, `memory_sources/rpc.rs`), and the
|
||||
UnifiedMemory writeback via a new **`SkillDocSink`** seam trait. MCP transport stays host.
|
||||
- **Process glue:** `memory/global.rs` singleton + queue worker; `memory/source_scope.rs` task-locals; `memory/chat.rs`; embeddings provider wiring. *(Amended, plan §8 / W-EMB: the provider **implementations** in `src/openhuman/embeddings/` migrate upstream into `tinyagents::harness::embeddings` — trait gains `name`/`model_id`/`signature` byte-pinned to `provider={name};model={model};dims={dims}` (P10) — and tinycortex bridges `EmbeddingBackend` to that trait; the host keeps factory/config/RPC wiring only.)*
|
||||
- **Policy/UX:** `preferences.rs`, `remember.rs`, `tree_policy.rs`, `util/redact.rs`, config mapping.
|
||||
- **Host-retained `UnifiedMemory` namespace-document tier** (0.3 key finding) — the 10 tables that
|
||||
coexist in the shared DB but **do not move**: `memory_docs`, `graph_global`, `graph_namespace`,
|
||||
@@ -110,11 +124,23 @@ verification below were validated **locally** against the branch to prove the ba
|
||||
|
||||
### The adapter seam: `src/openhuman/tinycortex/` (W1, mirrors `src/openhuman/tinyagents/`)
|
||||
|
||||
**W1 seam files** (all against seam traits already present in the crate, §0.2):
|
||||
`embeddings.rs` (`EmbeddingBackend`/`Embedder`), `chat.rs` (`ChatProvider`/`Summariser`×2/
|
||||
`EntityExtractor`/`GoalsGenerator`), `queue_driver.rs` (`QueueDelegates` + tokio worker loop +
|
||||
Sentry/bus), `config.rs` (`Config`→`MemoryConfig`), `sinks.rs` (`TreeJobSink`/`TreeLeafSink`/
|
||||
`SnapshotItemSource`/`EntityOccurrenceIndex`), `bus.rs` (engine outcomes → `DomainEvent`),
|
||||
`mod.rs` (facade re-exports + boundary doc). All 17 seam traits confirmed present (§0.2).
|
||||
`mod.rs` (facade re-exports + boundary doc). All 17 W1 seam traits confirmed present (§0.2).
|
||||
|
||||
**Later seam additions (amended 2026-07-09):**
|
||||
|
||||
- **W-EMB rebridge** — `embeddings.rs` is re-pointed from `openhuman::embeddings` to
|
||||
`tinyagents::harness::embeddings::EmbeddingModel` (crate bridges `EmbeddingBackend` to that trait,
|
||||
plan §8.2 / gap-audit roster). Seam file stays; its backing implementation moves upstream.
|
||||
- **W-SYNC seam file** — `sync_sink.rs` adds the host adapters for **two new crate traits** landing
|
||||
with W-SYNC.1 (not among the 17 above; they do not exist in the crate at the Phase-0 SHA):
|
||||
`SyncEventSink` (→ `MemorySyncStage` bus events) and `SkillDocSink` (→
|
||||
`MemoryClient::store_skill_sync`, the host-retained namespace-document tier). See plan §8.1 and the
|
||||
gap-audit roster.
|
||||
|
||||
---
|
||||
|
||||
@@ -135,21 +161,26 @@ audit SHA.
|
||||
| `memory_graph/` | 3 (0) | W7 | gap **G2** resolved (derive-on-read parity vs host-retained `graph_*`) |
|
||||
| `memory_goals/` | 7 (0) | W7 | seam `GoalsGenerator` wired |
|
||||
| `memory_archivist/` | 6 (0) | W7 | `TreeLeafSink` seam wired |
|
||||
| `memory_sources/` | 16 (0) | W7 | registry + local readers move; **live sync stays host** |
|
||||
| `memory_sources/` | 16 (0) | W7 / W-SYNC | registry + local readers move (W7); `sync.rs` dispatcher + `reconcile.rs` move in W-SYNC; `rpc.rs` kept host |
|
||||
| `memory_sync/` (engine) | — | **W-SYNC.3** | drift **D4** closed; W6 landed (crate ingest live); mocked + live Composio test pair green; sync-status parity green; schedulers/bus/RPC/keychain kept host |
|
||||
| `src/openhuman/embeddings/` (provider impls) | — | **W-EMB.3** | tinyagents provider port merged; signature parity (P10) green; `factory.rs`(thin)/`rpc.rs`/`schemas.rs`/catalog kept host |
|
||||
| `memory_tools/` | 10 (1) | W7 | engine → `tool_memory/`; tool surface kept host |
|
||||
| `memory_search/` | 8 (0) | W5 | `vector`/`scoring` → crate `retrieval`/`score`; `tools/` kept host |
|
||||
| `memory/ingest_pipeline.rs` internals | (thin entry points kept) | W6 | `ingest_chat`/`ingest_document_with_scope` signatures unchanged; 11 call sites untouched |
|
||||
|
||||
**Kept host (never deleted):** `memory/{ops,schemas,schema,read_rpc,tools,query,tree_source,
|
||||
ingestion,util}`, `memory/{global,source_scope,chat,sync,preferences,remember,tree_policy,rpc_models,
|
||||
traits(→re-exports)}.rs`, all of `memory_sync/`, `memory_store/unified/*` (the namespace-document
|
||||
traits(→re-exports)}.rs`, `memory_sync/`'s host-retained shell only (schedulers `periodic.rs`,
|
||||
`bus.rs` subscribers, RPC registration — the engine moves in W-SYNC, plan §8), `memory_store/unified/*` (the namespace-document
|
||||
tier), `memory_store/content/{wiki_git,obsidian,obsidian_registry}`, `memory_tree/health/`, and the
|
||||
new `src/openhuman/tinycortex/` seam.
|
||||
|
||||
## 3. Workstream order (one workstream ≈ one host PR)
|
||||
|
||||
W1 seam scaffolding → W2 types/trait re-export → W3 store+chunks → W4 queue → W5 tree+retrieval+score
|
||||
→ W6 ingest → W7 long tail → W8 test-port + golden parity sweep + deletion-ledger close-out.
|
||||
→ W6 ingest → **W-SYNC** (sync engine + Composio client, plan §8) → W7 long tail → W8 test-port +
|
||||
golden parity sweep + deletion-ledger close-out. **W-EMB** (embeddings inheritance from tinyagents,
|
||||
plan §8.2) runs in parallel; W-EMB.2 must land before the W-SYNC.3 flip.
|
||||
|
||||
Each risky workstream is a sandwich (plan §4): (a) tinycortex PR(s) closing that module's drift/gap
|
||||
ledger, (b) host `chore(vendor): bump tinycortex`, (c) host cutover PR (adapter flip + legacy
|
||||
@@ -161,3 +192,10 @@ deletion + host-side tests in the same PR for the ≥80% diff-coverage gate).
|
||||
2. **`source_scope` per-turn allowlist** — must survive the W5 retrieval cutover; the retrieval
|
||||
primitives (`query_source`/`query_topic`/`drill_down`) run inside the host's task-local scope, and
|
||||
a W5 seam test must assert an out-of-allowlist source is not returned.
|
||||
3. **Composio credential handling (W-SYNC)** — the crate holds the key only as a redacted
|
||||
`SecretString` (`Debug`/`Display` masked, serialization skipped); production resolution stays in
|
||||
the host keychain (`composio-direct`), the `COMPOSIO_API_KEY` env fallback is test-only; a seam
|
||||
test asserts the key never appears in `MemoryConfig` `Debug` output or client error messages
|
||||
(401 path included).
|
||||
4. **Sync taint provenance (W-SYNC)** — every crate-side sync ingest must stamp
|
||||
`MemoryTaint::ExternalSync`; the mocked Composio pipeline test pins this.
|
||||
|
||||
@@ -62,24 +62,41 @@ two layers.
|
||||
|
||||
### Layer 1 — schema/format asserters (host-side unit tests, cheap, run every PR)
|
||||
|
||||
A `tests/tinycortex_parity/` module with pure-function comparators (no disk):
|
||||
Pure-function comparators (no disk), implemented in **`src/openhuman/tinycortex/parity.rs`**
|
||||
(`#[cfg(test)]`). Status ✅ = landed & green; ⏳ = pending.
|
||||
|
||||
- **`chunk_id_parity`** — table of `(source_kind, source_id, seq, content)` → assert host `chunk_id`
|
||||
== `tinycortex::memory::chunks::chunk_id` (covers P1). *(After W3 both resolve to the crate; keep
|
||||
the vector as a regression pin.)*
|
||||
- **`vector_roundtrip_parity`** — random `Vec<f32>` → `vec_to_bytes` → `bytes_to_vec` byte-equal
|
||||
across both (P2).
|
||||
- **`content_path_parity`** — corpus of adversarial ids (colons, unicode, >255 chars, `email`
|
||||
source) → assert identical `chunk_rel_path`/`summary_rel_path` (P6).
|
||||
- **`frontmatter_parity`** — compose a fixed chunk+summary → byte-compare markdown incl. frontmatter
|
||||
key order (P7).
|
||||
- **`embedding_signature_parity`** — assert the W1 seam's `signature()` string == the format the
|
||||
store persisted into `store_meta` (P10).
|
||||
- ✅ **`chunk_id_matches_historical_golden` / `chunk_id_is_sensitive_to_every_field`** — golden +
|
||||
every-field sensitivity for the deterministic `chunk_id` (covers P1). *(After W3 both resolve to the
|
||||
crate; the golden vector stays as a regression pin.)*
|
||||
- ✅ **`vector_encoding_is_le_packed_f32`** — `vec_to_bytes`/`bytes_to_vec` LE-packed-f32 round-trip
|
||||
+ golden bytes (P2).
|
||||
- ✅ **`chunk_rel_path_host_crate_byte_parity` / `summary_rel_path_host_crate_byte_parity`** —
|
||||
adversarial id corpus (colons, all Windows-illegal chars, unicode, >255 chars, gmail participant
|
||||
slugs, malformed email; every summary-id shape × 3 tree kinds × levels) → assert host
|
||||
`chunk_rel_path`/`summary_rel_path` **byte-equal** the crate's (P6). *(Landed; verified identical.)*
|
||||
- ✅ **`embedding_signature_host_crate_byte_parity`** — assert host
|
||||
`embeddings::format_embedding_signature` == crate `store::vectors::format_embedding_signature` and
|
||||
both == the golden `provider={name};model={model};dims={dims}` over a provider corpus (P10). The
|
||||
seam's own `signature()` pass-through is separately pinned in `tinycortex/embeddings.rs`.
|
||||
- ⏳ **`frontmatter_parity`** — compose a fixed chunk+summary → byte-compare markdown incl.
|
||||
frontmatter key order (P7). *(Not yet landed — needs the host/crate compose types aligned.)*
|
||||
|
||||
### Layer 2 — golden-workspace differential harness (the flip gate)
|
||||
|
||||
The core mechanism from plan §0.3: **one on-disk workspace, opened by both engines, outputs compared.**
|
||||
|
||||
> **Status (2026-07-10).** First cut landed as **`tests/memory_golden_parity_e2e.rs`** —
|
||||
> comparator **1** (schema composition) and comparator **5** (idempotent re-open) are green:
|
||||
> a real workspace is stood up through the host production surface (`memory::ops`), the crate
|
||||
> substrate init is forced deterministically, and **all `*.db` files under the workspace are scanned
|
||||
> path-agnostically** (union of tables). It asserts the crate chunk-DB substrate (15 `chunks/schema.rs`
|
||||
> tables) and the host `UnifiedMemory` tier (10 tables) **coexist without collision** (P3/P5/P11/P12),
|
||||
> and that re-running the flow adds/drops no tables (comparator 5). Still TODO: `vectors`/`store_meta`/
|
||||
> `kv_*` (created by the chunk/embed pipeline, need a widened ingest flow), the seeded golden fixture +
|
||||
> `scripts/gen-golden-workspace.sh`, and comparators **2** (recall/retrieval snapshot), **3** (tree
|
||||
> read), **4** (byte-compare vault) — these require a populated, sealed fixture and the W5 retrieval
|
||||
> surface, so they land alongside the W3/W5 flips.
|
||||
|
||||
**Fixture.** Check in a small, deterministic `tests/fixtures/golden-workspace/` produced by the
|
||||
*pre-migration* build: a real `chunks.db` + content vault + diff `.git`, seeded via a fixed script
|
||||
(`scripts/gen-golden-workspace.sh`) with: a handful of chat + document + email sources across ≥2
|
||||
@@ -115,6 +132,19 @@ flip preserves *existing* data, not just that the API still functions.
|
||||
**W3** (store+chunks), **W5** (tree+retrieval+score), **W6** (ingest). W4 (queue) additionally asserts
|
||||
job payload_json parity (P4/P9). Any red = upstream fix in tinycortex, re-bump submodule, re-run.
|
||||
|
||||
**W-SYNC gates (amendment 2026-07-09, plan §8):**
|
||||
- **P13 sync-status parity** — `memory_sync_status_list` output (per-`source_kind` freshness rows)
|
||||
byte-equal pre/post flip on a golden workspace; asserter added to
|
||||
`src/openhuman/tinycortex/parity.rs`.
|
||||
- **P14 Composio sync test pair** — the crate's mocked-HTTP gmail pipeline test
|
||||
(`vendor/tinycortex/tests/composio_sync_mock.rs`, wiremock, always-on) green in crate CI before
|
||||
the W-SYNC.3 flip; the live `#[ignore]` test (`composio_sync_live.rs`, `COMPOSIO_API_KEY`) run
|
||||
manually at least once per flip wave.
|
||||
|
||||
**W-EMB gate:** the existing **P10 `embedding_signature_parity`** asserter is the regression pin —
|
||||
the tinyagents-backed provider stack must emit byte-identical
|
||||
`provider={name};model={model};dims={dims}` signatures, or existing vector spaces split.
|
||||
|
||||
## Open divergences to resolve upstream before their flip
|
||||
|
||||
| Item | Flip gated | Resolution |
|
||||
|
||||
@@ -32,5 +32,6 @@ scheduler (1 task) → daily wall-clock tick → `digest_daily(yesterday)`
|
||||
- `store.rs` — SQLite persistence: `INSERT OR IGNORE` + partial unique index on `dedupe_key WHERE status IN ('ready','running')` for at-most-one-active dedupe; `claim_next` is a single `UPDATE ... RETURNING`; `mark_done`/`mark_failed` are claim-token gated to make stale-worker settlements no-ops.
|
||||
- `worker.rs` — three worker tasks plus startup `recover_stale_locks` and a 3-permit semaphore around LLM-bound jobs. Calls into `crate::openhuman::scheduler_gate::wait_for_capacity()` before claiming so Throttled / Paused modes back off without holding DB leases.
|
||||
- `scheduler.rs` — daily tick at UTC 00:05 that enqueues `digest_daily(yesterday)` + `flush_stale(today)`; `trigger_digest` and `backfill_missing_digests` are manual catch-up helpers.
|
||||
- `handlers/` — per-`JobKind` handler implementations.
|
||||
- `testing.rs` — `drain_until_idle` for tests that need the pipeline to settle synchronously.
|
||||
|
||||
Per-`JobKind` dispatch (the former `handlers/` module) was deleted at the W4 flip: `worker::run_once` now delegates claim → dispatch → settle to `tinycortex::memory::queue::run_once` through `crate::openhuman::tinycortex::HostQueueDelegates`, which bridges each heavy step back to the host `memory_tree`/score/embed engine.
|
||||
|
||||
@@ -1,20 +0,0 @@
|
||||
# Memory tree — jobs handlers
|
||||
|
||||
Per-`JobKind` handler implementations dispatched by `worker::run_once_with_semaphore`. Each handler parses its payload, performs side effects, and enqueues any follow-up work (typically inside the same SQLite transaction as its primary write so a crash doesn't lose downstream jobs).
|
||||
|
||||
## Public surface
|
||||
|
||||
- `pub async fn handle_job(config, job)` — `mod.rs` — branches on `job.kind` and invokes the matching handler.
|
||||
|
||||
## Handlers (private to the module)
|
||||
|
||||
- `handle_extract` — runs the scorer + LLM extractor over one chunk, packs the embedding, writes `mem_tree_score` + entity-index rows + chunk lifecycle in one tx, and enqueues the follow-up `append_buffer` and `topic_route` jobs. Also rewrites Obsidian-style `tags:` in the on-disk chunk markdown (best-effort, post-tx).
|
||||
- `handle_append_buffer` — hydrates a `LeafRef` (chunk or summary), pushes into the target tree's L0 buffer, and enqueues a `seal` job if the buffer crosses its budget. Updates chunk lifecycle (`buffered`) for source-tree appends. All in one tx.
|
||||
- `handle_seal` — seals exactly one buffer level via `bucket_seal::seal_one_level` (which atomically inserts the parent-cascade seal and summary-side `topic_route` for source trees). Topic-tree seals are sinks and do not enqueue further routing. Rewrites tags on the sealed summary's `.md` post-commit.
|
||||
- `handle_topic_route` — for each canonical entity associated with the node, asks the topic curator whether to spawn a topic tree, and enqueues an `append_buffer` per matched topic tree.
|
||||
- `handle_digest_daily` — invokes `tree_global::digest::end_of_day_digest` for the requested UTC date; idempotent via the digest's own `find_existing_daily` check.
|
||||
- `handle_flush_stale` — walks `list_stale_buffers` and enqueues a forced `seal` per buffer over the configured `DEFAULT_FLUSH_AGE_SECS` cap.
|
||||
|
||||
## Files
|
||||
|
||||
- `mod.rs` — `handle_job` dispatch and all handler bodies.
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,744 +0,0 @@
|
||||
use super::*;
|
||||
use crate::openhuman::memory::tree_source::registry::get_or_create_source_tree;
|
||||
use crate::openhuman::memory_queue::store::{count_by_status, count_total};
|
||||
use crate::openhuman::memory_queue::types::JobStatus;
|
||||
use crate::openhuman::memory_store::chunks::store::with_connection;
|
||||
use crate::openhuman::memory_store::content as content_store;
|
||||
use crate::openhuman::memory_tree::tree::bucket_seal::{append_leaf_deferred, LeafRef};
|
||||
use crate::openhuman::memory_tree::tree::store as src_store;
|
||||
use chrono::TimeZone;
|
||||
use rusqlite::params;
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn test_config() -> (TempDir, Config) {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let mut cfg = Config::default();
|
||||
cfg.workspace_dir = tmp.path().to_path_buf();
|
||||
cfg.memory_tree.embedding_endpoint = None;
|
||||
cfg.memory_tree.embedding_model = None;
|
||||
cfg.memory_tree.embedding_strict = false;
|
||||
(tmp, cfg)
|
||||
}
|
||||
|
||||
/// Build a minimal `Job` row for direct handler invocation. Mirrors
|
||||
/// what `claim_next` would produce for a freshly-claimed row.
|
||||
fn mk_running_job(kind: JobKind, payload_json: String) -> Job {
|
||||
let now_ms = chrono::Utc::now().timestamp_millis();
|
||||
Job {
|
||||
id: "test-job-id".into(),
|
||||
kind,
|
||||
payload_json,
|
||||
dedupe_key: None,
|
||||
status: JobStatus::Running,
|
||||
attempts: 1,
|
||||
max_attempts: 5,
|
||||
available_at_ms: now_ms,
|
||||
locked_until_ms: Some(now_ms + 60_000),
|
||||
last_error: None,
|
||||
created_at_ms: now_ms,
|
||||
started_at_ms: Some(now_ms),
|
||||
completed_at_ms: None,
|
||||
failure_reason: None,
|
||||
failure_class: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Count rows in `mem_tree_jobs` matching a specific kind.
|
||||
fn count_jobs_of_kind(cfg: &Config, kind: &str) -> u64 {
|
||||
with_connection(cfg, |conn| {
|
||||
let n: i64 = conn.query_row(
|
||||
"SELECT COUNT(*) FROM mem_tree_jobs WHERE kind = ?1",
|
||||
params![kind],
|
||||
|r| r.get(0),
|
||||
)?;
|
||||
Ok(n.max(0) as u64)
|
||||
})
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// Seed a source tree and push enough labeled leaves into its L0 buffer
|
||||
/// to cross `INPUT_TOKEN_BUDGET`, returning the tree. The caller can then
|
||||
/// fire `handle_seal` and inspect the result.
|
||||
async fn seed_source_tree_ready_to_seal(
|
||||
cfg: &Config,
|
||||
) -> crate::openhuman::memory_store::trees::types::Tree {
|
||||
use crate::openhuman::memory_store::chunks::store::upsert_chunks;
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
chunk_id, Chunk, Metadata, SourceKind, SourceRef,
|
||||
};
|
||||
let tree = get_or_create_source_tree(cfg, "slack:#eng").unwrap();
|
||||
let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
let chunk = Chunk {
|
||||
id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "handler-seed"),
|
||||
content: "alice@example.com leading the rollout".into(),
|
||||
metadata: Metadata {
|
||||
source_kind: SourceKind::Chat,
|
||||
source_id: "slack:#eng".into(),
|
||||
owner: "alice".into(),
|
||||
timestamp: ts,
|
||||
time_range: (ts, ts),
|
||||
tags: vec![],
|
||||
source_ref: Some(SourceRef::new("slack://x")),
|
||||
path_scope: None,
|
||||
},
|
||||
// Bust budget so the L0 buffer is "ready" for seal.
|
||||
token_count: 60_000,
|
||||
seq_in_source: 0,
|
||||
created_at: ts,
|
||||
partial_message: false,
|
||||
};
|
||||
upsert_chunks(cfg, &[chunk.clone()]).unwrap();
|
||||
// Stage to disk so `hydrate_leaf_inputs` can read the full body via
|
||||
// `read_chunk_body` when `handle_seal` fires and calls `seal_one_level`.
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap();
|
||||
with_connection(cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
crate::openhuman::memory_store::chunks::store::upsert_staged_chunks_tx(&tx, &staged)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
let leaf = LeafRef {
|
||||
chunk_id: chunk.id,
|
||||
token_count: 60_000,
|
||||
timestamp: ts,
|
||||
content: chunk.content,
|
||||
entities: vec![],
|
||||
topics: vec![],
|
||||
score: 0.5,
|
||||
};
|
||||
// append_leaf_deferred only buffers; doesn't seal. handle_seal will.
|
||||
let _ = append_leaf_deferred(cfg, &tree, &leaf).unwrap();
|
||||
tree
|
||||
}
|
||||
|
||||
/// #1574 §6: a chunk with content but no sidecar vector at the active
|
||||
/// signature (the post-switch / dim-mismatch state) is re-embedded by
|
||||
/// `handle_reembed_backfill`; the chain `Defer`s while work remains and
|
||||
/// returns `Done` once the space is covered; a stale-signature job
|
||||
/// finishes immediately without touching anything.
|
||||
///
|
||||
/// (The process-global `backfill_in_progress` flag is intentionally not
|
||||
/// asserted here — it is shared across parallel tests and set widely by
|
||||
/// the §7 trigger, so asserting it would be flaky. The handler's
|
||||
/// deterministic effects are what this test pins.)
|
||||
#[tokio::test]
|
||||
async fn reembed_backfill_repopulates_then_completes() {
|
||||
use crate::openhuman::memory_store::chunks::store::{
|
||||
get_chunk_embedding_for_signature, tree_active_signature, upsert_chunks,
|
||||
upsert_staged_chunks_tx,
|
||||
};
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
chunk_id, Chunk, Metadata, SourceKind, SourceRef,
|
||||
};
|
||||
|
||||
let (_tmp, mut cfg) = test_config();
|
||||
// Deliberate "none" opt-out → build_write_embedder yields an InertEmbedder
|
||||
// (correct-dim zero vectors, no network). This test pins backfill
|
||||
// *mechanics* (worklist → sidecar write → Defer/Done), not embed quality;
|
||||
// the no-provider skip path is covered separately.
|
||||
cfg.embeddings_provider = Some("none".to_string());
|
||||
let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
let chunk = Chunk {
|
||||
id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "reembed-seed"),
|
||||
content: "memory content about the phoenix migration project".into(),
|
||||
metadata: Metadata {
|
||||
source_kind: SourceKind::Chat,
|
||||
source_id: "slack:#eng".into(),
|
||||
owner: "alice".into(),
|
||||
timestamp: ts,
|
||||
time_range: (ts, ts),
|
||||
tags: vec![],
|
||||
source_ref: Some(SourceRef::new("slack://x")),
|
||||
path_scope: None,
|
||||
},
|
||||
token_count: 12,
|
||||
seq_in_source: 0,
|
||||
created_at: ts,
|
||||
partial_message: false,
|
||||
};
|
||||
upsert_chunks(&cfg, &[chunk.clone()]).unwrap();
|
||||
// Stage the body to disk so `read_chunk_body` succeeds in the handler.
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap();
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
upsert_staged_chunks_tx(&tx, &staged)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let sig = tree_active_signature(&cfg);
|
||||
assert!(
|
||||
get_chunk_embedding_for_signature(&cfg, &chunk.id, &sig)
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"precondition: no sidecar vector at the active signature"
|
||||
);
|
||||
|
||||
// Work present → re-embed + write sidecar, Defer to revisit.
|
||||
let job = mk_running_job(
|
||||
JobKind::ReembedBackfill,
|
||||
serde_json::to_string(&ReembedBackfillPayload {
|
||||
signature: sig.clone(),
|
||||
})
|
||||
.unwrap(),
|
||||
);
|
||||
let out = handle_reembed_backfill(&cfg, &job).await.unwrap();
|
||||
assert!(
|
||||
matches!(out, JobOutcome::Defer { .. }),
|
||||
"work present must Defer (self-continue), got {out:?}"
|
||||
);
|
||||
assert!(
|
||||
get_chunk_embedding_for_signature(&cfg, &chunk.id, &sig)
|
||||
.unwrap()
|
||||
.is_some(),
|
||||
"chunk re-embedded into the sidecar at the active signature"
|
||||
);
|
||||
|
||||
// Nothing left → Done.
|
||||
let out2 = handle_reembed_backfill(&cfg, &job).await.unwrap();
|
||||
assert_eq!(out2, JobOutcome::Done, "covered space must complete");
|
||||
|
||||
// Stale signature (embedder changed since enqueue) → finishes
|
||||
// immediately, no work, no panic.
|
||||
let stale = mk_running_job(
|
||||
JobKind::ReembedBackfill,
|
||||
serde_json::to_string(&ReembedBackfillPayload {
|
||||
signature: "provider=other;model=x;dims=1".into(),
|
||||
})
|
||||
.unwrap(),
|
||||
);
|
||||
assert_eq!(
|
||||
handle_reembed_backfill(&cfg, &stale).await.unwrap(),
|
||||
JobOutcome::Done
|
||||
);
|
||||
}
|
||||
|
||||
/// #002 (FR-002) regression gate: when NO usable embeddings provider is
|
||||
/// configured, the re-embed backfill must SKIP (return `Done`) instead of
|
||||
/// falling back to an `InertEmbedder` and persisting all-zero vectors that
|
||||
/// would silently poison semantic recall — the same hazard the extract and
|
||||
/// seal write paths already guard against. The chunk stays embedding-less at
|
||||
/// the active signature (re-embeddable once a provider is configured).
|
||||
#[tokio::test]
|
||||
async fn reembed_backfill_skips_when_no_provider() {
|
||||
use crate::openhuman::memory_store::chunks::store::{
|
||||
get_chunk_embedding_for_signature, tree_active_signature, upsert_chunks,
|
||||
upsert_staged_chunks_tx,
|
||||
};
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
chunk_id, Chunk, Metadata, SourceKind, SourceRef,
|
||||
};
|
||||
|
||||
// Default test config leaves embeddings unconfigured (no endpoint/model,
|
||||
// provider unset) — the no-provider path build_write_embedder guards.
|
||||
//
|
||||
// Hold the shared health test-guard: the no-provider path marks the
|
||||
// process-global semantic-recall degraded flag, so the guard resets it on
|
||||
// entry and keeps the signal from leaking into parallel status tests.
|
||||
let _health_guard = crate::openhuman::memory_tree::health::test_guard();
|
||||
let (_tmp, cfg) = test_config();
|
||||
let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
let chunk = Chunk {
|
||||
id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "no-provider-seed"),
|
||||
content: "memory content with no embeddings provider configured".into(),
|
||||
metadata: Metadata {
|
||||
source_kind: SourceKind::Chat,
|
||||
source_id: "slack:#eng".into(),
|
||||
owner: "alice".into(),
|
||||
timestamp: ts,
|
||||
time_range: (ts, ts),
|
||||
tags: vec![],
|
||||
source_ref: Some(SourceRef::new("slack://x")),
|
||||
path_scope: None,
|
||||
},
|
||||
token_count: 12,
|
||||
seq_in_source: 0,
|
||||
created_at: ts,
|
||||
partial_message: false,
|
||||
};
|
||||
upsert_chunks(&cfg, &[chunk.clone()]).unwrap();
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap();
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
upsert_staged_chunks_tx(&tx, &staged)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let sig = tree_active_signature(&cfg);
|
||||
let job = mk_running_job(
|
||||
JobKind::ReembedBackfill,
|
||||
serde_json::to_string(&ReembedBackfillPayload {
|
||||
signature: sig.clone(),
|
||||
})
|
||||
.unwrap(),
|
||||
);
|
||||
|
||||
// No provider → skip the whole backfill (Done), do NOT write a vector.
|
||||
let out = handle_reembed_backfill(&cfg, &job).await.unwrap();
|
||||
assert_eq!(
|
||||
out,
|
||||
JobOutcome::Done,
|
||||
"no usable provider must skip the backfill, not Defer/embed"
|
||||
);
|
||||
assert!(
|
||||
get_chunk_embedding_for_signature(&cfg, &chunk.id, &sig)
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"no zero/inert vector may be persisted when no provider is configured"
|
||||
);
|
||||
}
|
||||
|
||||
/// #1574 §6 regression gate: a terminal-failure chunk (its body file is
|
||||
/// missing on disk, despite the metadata row staying staged) is
|
||||
/// persistently tombstoned by `mark_chunk_reembed_skipped` on the first
|
||||
/// pass, then excluded from the next batch's worklist so the chain
|
||||
/// terminates (`Done`) instead of looping forever. Without this guard
|
||||
/// the §6 runaway-loop fix would silently regress — the same 16 orphans
|
||||
/// → ~8k defers → ~128k warns symptom observed in the wild before the
|
||||
/// fix landed (see PR body and store.rs:1195).
|
||||
///
|
||||
/// What the test pins:
|
||||
/// 1. Tombstone row is written for the failing chunk (exactly one).
|
||||
/// 2. The next-batch worklist `NOT EXISTS … reembed_skipped` clause
|
||||
/// excludes the tombstoned row — the handler returns `Done`.
|
||||
/// 3. The `ensure_reembed_backfill` migration probe agrees the space
|
||||
/// is covered (or the chain would re-arm on every config save).
|
||||
#[tokio::test]
|
||||
async fn reembed_backfill_tombstones_orphan_and_terminates() {
|
||||
use crate::openhuman::memory_store::chunks::store::{
|
||||
get_chunk_content_path, get_chunk_embedding_for_signature, tree_active_signature,
|
||||
upsert_chunks, upsert_staged_chunks_tx,
|
||||
};
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
chunk_id, Chunk, Metadata, SourceKind, SourceRef,
|
||||
};
|
||||
|
||||
let (_tmp, mut cfg) = test_config();
|
||||
// Deliberate "none" opt-out → InertEmbedder (zero vectors, no network) so
|
||||
// the backfill reaches the orphan body-read and tombstones it; this test
|
||||
// pins the tombstone-and-terminate mechanics, not embed quality.
|
||||
cfg.embeddings_provider = Some("none".to_string());
|
||||
let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
let chunk = Chunk {
|
||||
id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "orphan-seed"),
|
||||
content: "memory content about the orphaned phoenix project".into(),
|
||||
metadata: Metadata {
|
||||
source_kind: SourceKind::Chat,
|
||||
source_id: "slack:#eng".into(),
|
||||
owner: "alice".into(),
|
||||
timestamp: ts,
|
||||
time_range: (ts, ts),
|
||||
tags: vec![],
|
||||
source_ref: Some(SourceRef::new("slack://x")),
|
||||
path_scope: None,
|
||||
},
|
||||
token_count: 12,
|
||||
seq_in_source: 0,
|
||||
created_at: ts,
|
||||
partial_message: false,
|
||||
};
|
||||
upsert_chunks(&cfg, &[chunk.clone()]).unwrap();
|
||||
|
||||
// Stage the body file + metadata, then DELETE the body file from
|
||||
// disk while leaving the staged DB rows intact. Reproduces the
|
||||
// in-wild failure mode: chunk row + path hash both present, but
|
||||
// the body content was lost (user moved workspace dirs, partial
|
||||
// backup restore, manual file cleanup). `stage_chunks` returns
|
||||
// paths relative to `content_root`; resolve absolute before unlink.
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap();
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
upsert_staged_chunks_tx(&tx, &staged)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
let staged_rel = get_chunk_content_path(&cfg, &chunk.id)
|
||||
.unwrap()
|
||||
.expect("staged body path");
|
||||
let body_abs = content_root.join(&staged_rel);
|
||||
std::fs::remove_file(&body_abs).unwrap();
|
||||
|
||||
let sig = tree_active_signature(&cfg);
|
||||
let job = mk_running_job(
|
||||
JobKind::ReembedBackfill,
|
||||
serde_json::to_string(&ReembedBackfillPayload {
|
||||
signature: sig.clone(),
|
||||
})
|
||||
.unwrap(),
|
||||
);
|
||||
|
||||
// Pass 1: worklist picks up the orphan, body read fails, tombstone
|
||||
// written, `Defer` to revisit (the handler doesn't distinguish
|
||||
// "all rows tombstoned" from "more rows pending" inside this batch).
|
||||
let out1 = handle_reembed_backfill(&cfg, &job).await.unwrap();
|
||||
assert!(
|
||||
matches!(out1, JobOutcome::Defer { .. }),
|
||||
"first pass should Defer after failing to read body, got {out1:?}"
|
||||
);
|
||||
assert!(
|
||||
get_chunk_embedding_for_signature(&cfg, &chunk.id, &sig)
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"orphan chunk must not have a sidecar vector after failure"
|
||||
);
|
||||
|
||||
// (1) Tombstone row exists for exactly this (chunk, sig).
|
||||
let tombstone_count: i64 = with_connection(&cfg, |conn| {
|
||||
Ok(conn.query_row(
|
||||
"SELECT COUNT(*) FROM mem_tree_chunk_reembed_skipped
|
||||
WHERE chunk_id = ?1 AND model_signature = ?2",
|
||||
params![chunk.id, sig],
|
||||
|r| r.get(0),
|
||||
)?)
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
tombstone_count, 1,
|
||||
"orphan chunk must be tombstoned exactly once"
|
||||
);
|
||||
|
||||
// (2) Pass 2: worklist NOT EXISTS clause excludes the tombstoned
|
||||
// row; both worklists empty; chain completes.
|
||||
let out2 = handle_reembed_backfill(&cfg, &job).await.unwrap();
|
||||
assert_eq!(
|
||||
out2,
|
||||
JobOutcome::Done,
|
||||
"tombstoned-only state must complete the chain"
|
||||
);
|
||||
|
||||
// (3) Migration probe in `ensure_reembed_backfill` must agree the
|
||||
// space is covered, otherwise the chain re-arms on every config
|
||||
// save and we're back to the original infinite-loop bug.
|
||||
let probe_uncovered = with_connection(&cfg, |conn| {
|
||||
Ok(chunk_store::has_uncovered_reembed_work(conn, &sig)?)
|
||||
})
|
||||
.unwrap();
|
||||
assert!(
|
||||
!probe_uncovered,
|
||||
"after tombstoning the only orphan, the ensure_reembed_backfill probe must report covered"
|
||||
);
|
||||
}
|
||||
|
||||
/// #2358: clearing a tombstone re-opens the row for the backfill worklist.
|
||||
#[tokio::test]
|
||||
async fn clear_chunk_reembed_skipped_reopens_worklist() {
|
||||
use crate::openhuman::memory_store::chunks::store::{
|
||||
clear_chunk_reembed_skipped, get_chunk_content_path, mark_chunk_reembed_skipped,
|
||||
tree_active_signature, upsert_chunks, upsert_staged_chunks_tx,
|
||||
};
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
chunk_id, Chunk, Metadata, SourceKind, SourceRef,
|
||||
};
|
||||
|
||||
let (_tmp, cfg) = test_config();
|
||||
let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
let chunk = Chunk {
|
||||
id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "clear-tombstone-seed"),
|
||||
content: "memory content for clear tombstone test".into(),
|
||||
metadata: Metadata {
|
||||
source_kind: SourceKind::Chat,
|
||||
source_id: "slack:#eng".into(),
|
||||
owner: "alice".into(),
|
||||
timestamp: ts,
|
||||
time_range: (ts, ts),
|
||||
tags: vec![],
|
||||
source_ref: Some(SourceRef::new("slack://x")),
|
||||
path_scope: None,
|
||||
},
|
||||
token_count: 12,
|
||||
seq_in_source: 0,
|
||||
created_at: ts,
|
||||
partial_message: false,
|
||||
};
|
||||
upsert_chunks(&cfg, &[chunk.clone()]).unwrap();
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap();
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
upsert_staged_chunks_tx(&tx, &staged)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
let staged_rel = get_chunk_content_path(&cfg, &chunk.id)
|
||||
.unwrap()
|
||||
.expect("staged body path");
|
||||
std::fs::remove_file(content_root.join(&staged_rel)).unwrap();
|
||||
|
||||
let sig = tree_active_signature(&cfg);
|
||||
mark_chunk_reembed_skipped(&cfg, &chunk.id, &sig, "orphan").unwrap();
|
||||
|
||||
let covered_before_clear = with_connection(&cfg, |conn| {
|
||||
Ok(!chunk_store::has_uncovered_reembed_work(conn, &sig)?)
|
||||
})
|
||||
.unwrap();
|
||||
assert!(
|
||||
covered_before_clear,
|
||||
"tombstone must hide orphan from uncovered probe"
|
||||
);
|
||||
|
||||
clear_chunk_reembed_skipped(&cfg, &chunk.id, &sig).unwrap();
|
||||
|
||||
let uncovered_after_clear = with_connection(&cfg, |conn| {
|
||||
Ok(chunk_store::has_uncovered_reembed_work(conn, &sig)?)
|
||||
})
|
||||
.unwrap();
|
||||
assert!(
|
||||
uncovered_after_clear,
|
||||
"clearing tombstone must re-include chunk in worklist probe"
|
||||
);
|
||||
}
|
||||
|
||||
/// #1574 §4: `ensure_reembed_backfill` (the switch-path trigger) enqueues
|
||||
/// exactly one chain when there is uncovered work, is idempotent on
|
||||
/// re-call (per-signature dedupe), and enqueues nothing for an
|
||||
/// empty/covered space.
|
||||
#[tokio::test]
|
||||
async fn ensure_reembed_backfill_enqueues_only_when_uncovered() {
|
||||
use crate::openhuman::memory_queue::ensure_reembed_backfill;
|
||||
use crate::openhuman::memory_store::chunks::store::{upsert_chunks, upsert_staged_chunks_tx};
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
chunk_id, Chunk, Metadata, SourceKind, SourceRef,
|
||||
};
|
||||
|
||||
// Empty space → nothing to do → no job.
|
||||
let (_t0, empty_cfg) = test_config();
|
||||
ensure_reembed_backfill(&empty_cfg);
|
||||
assert_eq!(
|
||||
count_jobs_of_kind(&empty_cfg, "reembed_backfill"),
|
||||
0,
|
||||
"empty/covered space must not enqueue a backfill"
|
||||
);
|
||||
|
||||
// Chunk with content but no sidecar vector → exactly one chain.
|
||||
let (_t1, cfg) = test_config();
|
||||
let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
let chunk = Chunk {
|
||||
id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "ensure-seed"),
|
||||
content: "memory content needing a re-embed".into(),
|
||||
metadata: Metadata {
|
||||
source_kind: SourceKind::Chat,
|
||||
source_id: "slack:#eng".into(),
|
||||
owner: "alice".into(),
|
||||
timestamp: ts,
|
||||
time_range: (ts, ts),
|
||||
tags: vec![],
|
||||
source_ref: Some(SourceRef::new("slack://x")),
|
||||
path_scope: None,
|
||||
},
|
||||
token_count: 12,
|
||||
seq_in_source: 0,
|
||||
created_at: ts,
|
||||
partial_message: false,
|
||||
};
|
||||
upsert_chunks(&cfg, &[chunk.clone()]).unwrap();
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap();
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
upsert_staged_chunks_tx(&tx, &staged)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
ensure_reembed_backfill(&cfg);
|
||||
assert_eq!(
|
||||
count_jobs_of_kind(&cfg, "reembed_backfill"),
|
||||
1,
|
||||
"uncovered work must enqueue exactly one backfill chain"
|
||||
);
|
||||
// Idempotent — re-call must not create a second chain (dedupe by sig).
|
||||
ensure_reembed_backfill(&cfg);
|
||||
assert_eq!(
|
||||
count_jobs_of_kind(&cfg, "reembed_backfill"),
|
||||
1,
|
||||
"re-call must dedupe to a single chain per signature"
|
||||
);
|
||||
}
|
||||
|
||||
/// `cap_embed_text` must bound an over-budget body to `EMBED_SAFE_TOKENS` before
|
||||
/// it is batched for embedding, and pass small bodies through verbatim, so no
|
||||
/// single `embed_batch` input can exceed the embedder's input limit.
|
||||
#[test]
|
||||
fn cap_embed_text_bounds_oversized_input() {
|
||||
use crate::openhuman::memory_store::chunks::types::conservative_token_estimate;
|
||||
|
||||
// Punctuation-dense body (~1 token/char) far over the embed budget.
|
||||
let big = "x,".repeat(EMBED_SAFE_TOKENS as usize);
|
||||
assert!(conservative_token_estimate(&big) > EMBED_SAFE_TOKENS);
|
||||
let capped = cap_embed_text(&big);
|
||||
assert!(capped.len() < big.len(), "oversized body must be truncated");
|
||||
assert!(
|
||||
conservative_token_estimate(capped) <= EMBED_SAFE_TOKENS,
|
||||
"capped body must be within the embed budget",
|
||||
);
|
||||
assert!(big.starts_with(capped), "cap must return a leading prefix");
|
||||
|
||||
// A small body is returned unchanged.
|
||||
let small = "hello world";
|
||||
assert_eq!(cap_embed_text(small), small);
|
||||
}
|
||||
|
||||
/// C9: `prepare_extract` must score over the **full** on-disk body but leave
|
||||
/// the returned `PreparedExtract.chunk` holding only the ≤500-char preview —
|
||||
/// never the full body. This is the swap-then-restore guarantee that keeps a
|
||||
/// batch of N prepared items from retaining N full bodies in memory before
|
||||
/// finalize. We assert the body is read from disk (a body longer than the
|
||||
/// preview) yet the returned chunk content equals the preview again.
|
||||
#[tokio::test]
|
||||
async fn prepare_extract_scores_full_body_but_returns_preview() {
|
||||
use crate::openhuman::memory::chat::{test_override, StaticChatProvider};
|
||||
use crate::openhuman::memory_queue::types::ExtractChunkPayload;
|
||||
use crate::openhuman::memory_store::chunks::store::upsert_chunks;
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
chunk_id, Chunk, Metadata, SourceKind, SourceRef,
|
||||
};
|
||||
use std::sync::Arc;
|
||||
|
||||
let (_tmp, cfg) = test_config();
|
||||
let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
|
||||
// Full body is > 500 chars so the stored preview is a strict prefix and
|
||||
// is observably different from the body the scorer reads off disk.
|
||||
let body: String = "the quarterly planning notes from the engineering sync covering rollout sequencing and staffing. "
|
||||
.repeat(12);
|
||||
assert!(body.len() > 500, "test body must exceed the preview cap");
|
||||
let expected_preview: String = body.chars().take(500).collect();
|
||||
|
||||
let chunk = Chunk {
|
||||
id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "c9-extract-seed"),
|
||||
content: body.clone(),
|
||||
metadata: Metadata {
|
||||
source_kind: SourceKind::Chat,
|
||||
source_id: "slack:#eng".into(),
|
||||
owner: "alice".into(),
|
||||
timestamp: ts,
|
||||
time_range: (ts, ts),
|
||||
tags: vec![],
|
||||
source_ref: Some(SourceRef::new("slack://x")),
|
||||
path_scope: None,
|
||||
},
|
||||
token_count: 64,
|
||||
seq_in_source: 0,
|
||||
created_at: ts,
|
||||
partial_message: false,
|
||||
};
|
||||
// upsert_chunks stores only the ≤500-char preview in the `content` column.
|
||||
upsert_chunks(&cfg, &[chunk.clone()]).unwrap();
|
||||
// Stage the full body to disk so `read_chunk_body` returns it.
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
std::fs::create_dir_all(&content_root).unwrap();
|
||||
let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap();
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
crate::openhuman::memory_store::chunks::store::upsert_staged_chunks_tx(&tx, &staged)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let payload = ExtractChunkPayload {
|
||||
chunk_id: chunk.id.clone(),
|
||||
};
|
||||
let job = mk_running_job(
|
||||
JobKind::ExtractChunk,
|
||||
serde_json::to_string(&payload).unwrap(),
|
||||
);
|
||||
|
||||
// Pin a static (offline) chat provider so any borderline LLM consult is
|
||||
// deterministic and never touches the network.
|
||||
let prepared = test_override::with_provider(
|
||||
Arc::new(StaticChatProvider::new("[]")),
|
||||
prepare_extract(&cfg, &job),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("seeded chunk must produce a PreparedExtract");
|
||||
|
||||
assert_eq!(
|
||||
prepared.chunk.content, expected_preview,
|
||||
"returned chunk must hold the restored preview, not the full body"
|
||||
);
|
||||
assert_ne!(
|
||||
prepared.chunk.content, body,
|
||||
"returned chunk must NOT retain the full on-disk body"
|
||||
);
|
||||
}
|
||||
|
||||
/// #4359: a cloud-embedding "No backend session" bail is a global, login-
|
||||
/// recoverable condition — `reembed_collect` must fail the backfill fast with a
|
||||
/// typed `AuthMissing` failure and must **not** tombstone any row, so the same
|
||||
/// rows re-embed once the user signs in. (A per-row tombstone here would
|
||||
/// exclude the rows from the worklist forever — the regression this guards.)
|
||||
struct MissingSessionEmbedder;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl crate::openhuman::memory_tree::score::embed::Embedder for MissingSessionEmbedder {
|
||||
fn name(&self) -> &'static str {
|
||||
"missing-session"
|
||||
}
|
||||
|
||||
async fn embed(&self, _text: &str) -> anyhow::Result<Vec<f32>> {
|
||||
anyhow::bail!(
|
||||
"No backend session for cloud embeddings: log in to OpenHuman, or set \
|
||||
memory.embedding_provider to \"ollama\" / \"none\" in config.toml"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reembed_backfill_auth_missing_fails_fast_without_tombstone() {
|
||||
use crate::openhuman::memory_tree::health::{FailureCode, PipelineFailure};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
let (_tmp, cfg) = test_config();
|
||||
let ids = vec![
|
||||
"chunk-a".to_string(),
|
||||
"chunk-b".to_string(),
|
||||
"chunk-c".to_string(),
|
||||
];
|
||||
let tombstoned = Arc::new(AtomicUsize::new(0));
|
||||
let tombstoned_for_mark = Arc::clone(&tombstoned);
|
||||
|
||||
let err = reembed_collect(
|
||||
&cfg,
|
||||
&MissingSessionEmbedder,
|
||||
"provider=cloud;model=embedding-v1;dims=1024",
|
||||
&ids,
|
||||
"chunk",
|
||||
|_cfg, id| Ok(format!("body for {id}")),
|
||||
move |_cfg, _id, _sig, _reason| {
|
||||
tombstoned_for_mark.fetch_add(1, Ordering::SeqCst);
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("missing cloud session must fail the backfill, not return Ok");
|
||||
|
||||
let failure = err
|
||||
.downcast_ref::<PipelineFailure>()
|
||||
.expect("auth-missing backfill error must carry a typed PipelineFailure");
|
||||
assert_eq!(failure.code, FailureCode::AuthMissing);
|
||||
assert!(failure.is_unrecoverable());
|
||||
assert_eq!(
|
||||
tombstoned.load(Ordering::SeqCst),
|
||||
0,
|
||||
"auth-missing must NOT tombstone any row — they must stay re-embeddable after login"
|
||||
);
|
||||
}
|
||||
@@ -30,7 +30,6 @@
|
||||
//! execution/runtime concern rather than a leaf of the memory policy API.
|
||||
//! `openhuman::memory` re-exports it as `memory::jobs` during the migration.
|
||||
|
||||
pub(crate) mod handlers;
|
||||
mod ops;
|
||||
mod redact;
|
||||
pub mod scheduler;
|
||||
|
||||
@@ -151,59 +151,6 @@ pub fn claim_next(config: &Config, lock_duration_ms: i64) -> Result<Option<Job>>
|
||||
})
|
||||
}
|
||||
|
||||
/// Claim additional ready `extract_chunk` jobs for the same worker tick.
|
||||
///
|
||||
/// The caller has already claimed one job through [`claim_next`]. This helper
|
||||
/// leases up to `limit` more extract jobs so the handler can batch the
|
||||
/// embedding sub-step without widening non-extract scheduling semantics.
|
||||
pub fn claim_ready_extract_batch(
|
||||
config: &Config,
|
||||
lock_duration_ms: i64,
|
||||
limit: usize,
|
||||
) -> Result<Vec<Job>> {
|
||||
if limit == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
with_connection(config, |conn| {
|
||||
let now_ms = Utc::now().timestamp_millis();
|
||||
let lock_until = now_ms.saturating_add(lock_duration_ms);
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"UPDATE mem_tree_jobs
|
||||
SET status = 'running',
|
||||
attempts = attempts + 1,
|
||||
started_at_ms = ?1,
|
||||
locked_until_ms = ?2,
|
||||
last_error = NULL
|
||||
WHERE id IN (
|
||||
SELECT id FROM mem_tree_jobs
|
||||
WHERE status = 'ready'
|
||||
AND kind = 'extract_chunk'
|
||||
AND available_at_ms <= ?1
|
||||
ORDER BY available_at_ms ASC
|
||||
LIMIT ?3
|
||||
)
|
||||
RETURNING id, kind, payload_json, dedupe_key, status, attempts,
|
||||
max_attempts, available_at_ms, locked_until_ms, last_error,
|
||||
created_at_ms, started_at_ms, completed_at_ms,
|
||||
failure_reason, failure_class",
|
||||
)
|
||||
.context("prepare claim_ready_extract_batch")?;
|
||||
let jobs = stmt
|
||||
.query_map(params![now_ms, lock_until, limit as i64], row_to_job)?
|
||||
.collect::<rusqlite::Result<Vec<_>>>()
|
||||
.context("query claim_ready_extract_batch")?;
|
||||
if !jobs.is_empty() {
|
||||
log::debug!(
|
||||
"[memory::jobs] claimed extract batch count={} lock_until_ms={}",
|
||||
jobs.len(),
|
||||
lock_until
|
||||
);
|
||||
}
|
||||
Ok(jobs)
|
||||
})
|
||||
}
|
||||
|
||||
/// Mark a claimed job as `done`. Clears the lock and stamps `completed_at_ms`.
|
||||
///
|
||||
/// The UPDATE is gated on `attempts` and `started_at_ms` matching the values
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
//! Worker pool: claims jobs from `mem_tree_jobs`, dispatches them through
|
||||
//! [`handlers::handle_job`], and settles the row.
|
||||
//! Worker pool: drives the crate queue engine (W4 flip). Each `run_once`
|
||||
//! delegates claim → dispatch → settle to `tinycortex::memory::queue::run_once`
|
||||
//! via [`crate::openhuman::tinycortex::HostQueueDelegates`]; the legacy host
|
||||
//! `handlers` engine that used to own dispatch was deleted at the flip.
|
||||
//!
|
||||
//! Concurrency control for LLM-bound work is delegated to
|
||||
//! [`crate::openhuman::scheduler_gate`] — its global single-slot
|
||||
@@ -16,15 +18,13 @@ use anyhow::Result;
|
||||
use tokio::sync::Notify;
|
||||
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::memory_queue::handlers;
|
||||
use crate::openhuman::memory_queue::redact::scrub_for_log;
|
||||
use crate::openhuman::memory_queue::store::{
|
||||
claim_next, claim_ready_extract_batch, mark_deferred, mark_done, mark_failed_typed,
|
||||
recover_stale_locks, release_running_locks, DEFAULT_LOCK_DURATION_MS,
|
||||
};
|
||||
use crate::openhuman::memory_queue::types::{Job, JobKind, JobOutcome};
|
||||
// W4 flip: `run_once` now delegates claim/dispatch/settle to the crate, so the
|
||||
// legacy `handlers`, per-job settle (`mark_*`/`scrub_for_log`), and claim
|
||||
// helpers are gone from this module. Only startup lock recovery + the loop's
|
||||
// storage-degraded signalling remain host.
|
||||
use crate::openhuman::memory_queue::store::{recover_stale_locks, release_running_locks};
|
||||
use crate::openhuman::memory_tree::health::{
|
||||
clear_storage_degraded, mark_storage_degraded, FailureCode, PipelineFailure,
|
||||
clear_storage_degraded, mark_storage_degraded, FailureCode,
|
||||
};
|
||||
|
||||
/// Number of concurrent job-worker tasks. Each worker claims one job
|
||||
@@ -269,145 +269,24 @@ pub fn start(config: Config) {
|
||||
/// Claim and run a single job. Returns `true` when work was processed,
|
||||
/// `false` when no eligible row was available.
|
||||
pub async fn run_once(config: &Config) -> Result<bool> {
|
||||
// Cooperative throttle BEFORE `claim_next()`. Holding the DB claim
|
||||
// across an awaited `wait_for_capacity()` would let `Paused` mode
|
||||
// sit on the row past `DEFAULT_LOCK_DURATION_MS`, after which
|
||||
// `recover_stale_locks()` would requeue it for another worker to
|
||||
// pick up — duplicating side effects. Throttling here means
|
||||
// non-LLM jobs (AppendBuffer/FlushStale) also experience the same
|
||||
// gate delay, but that's fine: in Throttled mode the host is
|
||||
// already overloaded and a 30s breather between any DB-write batch
|
||||
// is welcome; in Paused mode the user has explicitly asked us to
|
||||
// stand down. Returns immediately in Aggressive/Normal so plugged-in
|
||||
// desktops with headroom pay zero cost.
|
||||
//
|
||||
// For LLM-bound jobs the returned `LlmPermit` reserves the global
|
||||
// single slot for the lifetime of `handle_job`. Non-LLM jobs
|
||||
// (`AppendBuffer`, `FlushStale`) drop the permit before the
|
||||
// handler runs so they don't block the slot.
|
||||
let gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await;
|
||||
// Cooperative throttle BEFORE claiming, so memory queue work still yields to
|
||||
// voice/autocomplete/triage under load (Throttled/Paused modes), exactly as
|
||||
// the legacy pool did. Held across the single crate step below; returns
|
||||
// immediately in Aggressive/Normal so idle desktops pay zero cost.
|
||||
let _gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await;
|
||||
|
||||
let Some(job) = claim_next(config, DEFAULT_LOCK_DURATION_MS)? else {
|
||||
return Ok(false);
|
||||
};
|
||||
|
||||
let llm_permit = if job.kind.is_llm_bound() {
|
||||
// Local Ollama loads ~1.3 GB resident per concurrent call —
|
||||
// hold the gate to enforce process-wide single-slot RAM
|
||||
// safety. Cloud calls are bandwidth-bound, not RAM-bound:
|
||||
// drop the permit so multiple workers can run cloud
|
||||
// extract/summarise calls in parallel (the worker pool
|
||||
// itself, sized to `WORKER_COUNT`, is the upstream bound).
|
||||
let memory_uses_local = config.workload_uses_local("memory");
|
||||
log::trace!(
|
||||
"[memory::jobs] llm permit routing job_id={} kind={} memory_uses_local={}",
|
||||
job.id,
|
||||
job.kind.as_str(),
|
||||
memory_uses_local
|
||||
);
|
||||
if memory_uses_local {
|
||||
gate_permit
|
||||
} else {
|
||||
drop(gate_permit);
|
||||
None
|
||||
}
|
||||
} else {
|
||||
// Non-LLM jobs don't need the global slot; release it so an
|
||||
// LLM-bound caller waiting elsewhere in the process can run.
|
||||
drop(gate_permit);
|
||||
None
|
||||
};
|
||||
|
||||
let mut jobs = vec![job];
|
||||
if jobs[0].kind == JobKind::ExtractChunk {
|
||||
let extra_limit = handlers::EXTRACT_EMBED_BATCH.saturating_sub(1);
|
||||
let mut extra = claim_ready_extract_batch(config, DEFAULT_LOCK_DURATION_MS, extra_limit)?;
|
||||
if !extra.is_empty() {
|
||||
log::debug!(
|
||||
"[memory::jobs] running extract batch count={}",
|
||||
extra.len() + 1
|
||||
);
|
||||
jobs.append(&mut extra);
|
||||
}
|
||||
}
|
||||
|
||||
let results = if jobs.len() > 1 && jobs[0].kind == JobKind::ExtractChunk {
|
||||
handlers::handle_extract_batch(config, &jobs).await?
|
||||
} else {
|
||||
let job = jobs
|
||||
.pop()
|
||||
.expect("worker has exactly one claimed job in non-batch path");
|
||||
let result = handlers::handle_job(config, &job).await;
|
||||
vec![(job, result)]
|
||||
};
|
||||
drop(llm_permit);
|
||||
|
||||
// A failed settle (`mark_done` / `mark_failed` / `mark_deferred` below)
|
||||
// can also return `SQLITE_BUSY`. The worker's outer `Err` arm in
|
||||
// `start` reclassifies those into a warn-log + backoff (no Sentry
|
||||
// report) via [`is_sqlite_busy`]. On a stale settle the row's
|
||||
// `locked_until_ms` eventually elapses and `recover_stale_locks`
|
||||
// requeues it, so dropping the error here is at-most a re-run.
|
||||
for (job, result) in results {
|
||||
settle_job(config, &job, result)?;
|
||||
}
|
||||
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
fn settle_job(config: &Config, job: &Job, result: Result<JobOutcome>) -> Result<()> {
|
||||
match result {
|
||||
Ok(JobOutcome::Done) => {
|
||||
log::debug!(
|
||||
"[memory::jobs] done id={} kind={}",
|
||||
job.id,
|
||||
job.kind.as_str()
|
||||
);
|
||||
mark_done(config, job)?;
|
||||
}
|
||||
Ok(JobOutcome::Defer { until_ms, reason }) => {
|
||||
// Defer is normal operation (transient blocker, e.g. rate
|
||||
// limit) — log at info, not warn — and do NOT count this
|
||||
// claim toward the failure-attempt budget. `mark_deferred`
|
||||
// reverts the bump applied by `claim_next` so the row's
|
||||
// attempts counter stays where it was before this claim.
|
||||
//
|
||||
// `reason` is handler-supplied free-form text and may
|
||||
// include upstream provider responses; scrub for log
|
||||
// emission while keeping the original in DB state.
|
||||
log::info!(
|
||||
"[memory::jobs] deferred id={} kind={} until_ms={} reason={}",
|
||||
job.id,
|
||||
job.kind.as_str(),
|
||||
until_ms,
|
||||
scrub_for_log(&reason)
|
||||
);
|
||||
mark_deferred(config, job, until_ms, &reason)?;
|
||||
}
|
||||
Err(err) => {
|
||||
// Preserve the full anyhow cause chain in the persisted
|
||||
// last_error so a reader of mem_tree_jobs can see the root
|
||||
// cause, not just the top-level message. The log line gets
|
||||
// the same chain after `scrub_for_log`, since anyhow chains
|
||||
// commonly embed upstream HTTP bodies / auth headers.
|
||||
let message = format!("{err:#}");
|
||||
// #002: if the error chain carries a typed `PipelineFailure`
|
||||
// (attached at the embed/extract boundary), pass it through so
|
||||
// `mark_failed_typed` can fail fast on unrecoverable causes
|
||||
// (budget/auth/dim) instead of burning the retry budget, and
|
||||
// persist the typed reason for the status/doctor surface.
|
||||
let typed = err.downcast_ref::<PipelineFailure>();
|
||||
log::warn!(
|
||||
"[memory::jobs] job failed id={} kind={} reason={:?} err={}",
|
||||
job.id,
|
||||
job.kind.as_str(),
|
||||
typed.map(|f| f.code.as_str()),
|
||||
scrub_for_log(&message)
|
||||
);
|
||||
mark_failed_typed(config, job, &message, typed)?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
// W4 flip: TinyCortex now owns claim → dispatch → settle. `queue::run_once`
|
||||
// claims one `mem_tree_jobs` row (the same table host producers enqueue
|
||||
// into — identical schema, parity P4) and runs it through the crate's
|
||||
// `handle_job` + `HostQueueDelegates`, which bridge each heavy step
|
||||
// (score/admit, buffer push, seal, seal-document, re-embed) back to the host
|
||||
// `memory_tree`/score/embed engine, then settles the row itself. The crate's
|
||||
// single-slot LLM gate serialises llm-bound jobs; the legacy per-job
|
||||
// local/cloud permit routing and the extract-batch coalescing are
|
||||
// intentionally dropped here (perf, not correctness — W4 follow-up).
|
||||
let mc = crate::openhuman::tinycortex::memory_config_from(config, config.workspace_dir.clone());
|
||||
let delegates = crate::openhuman::tinycortex::HostQueueDelegates::new(config.clone());
|
||||
tinycortex::memory::queue::run_once(&mc, &delegates).await
|
||||
}
|
||||
|
||||
/// Classify whether an error is a transient I/O failure that should be
|
||||
|
||||
@@ -554,21 +554,16 @@ async fn clear_memory_delete_cascades_orphaned_source_tree_and_settles_queued_jo
|
||||
.unwrap();
|
||||
|
||||
// ---- the queued Seal job settles to Done (tree missing), not stuck pending ----
|
||||
let claimed = queue_store::claim_next(&cfg, queue_store::DEFAULT_LOCK_DURATION_MS)
|
||||
.unwrap()
|
||||
.expect("seal job claimable");
|
||||
assert_eq!(claimed.kind, queue_types::JobKind::Seal);
|
||||
let outcome = crate::openhuman::memory_queue::handlers::handle_job(&cfg, &claimed)
|
||||
// Drive the crate queue engine end-to-end through the seam
|
||||
// (`drain_until_idle` → `worker::run_once` → crate claim/dispatch/settle),
|
||||
// the flipped W4 production path — rather than the deleted host engine.
|
||||
crate::openhuman::memory_queue::drain_until_idle(&cfg)
|
||||
.await
|
||||
.expect("handle_job ok");
|
||||
assert!(
|
||||
matches!(outcome, queue_types::JobOutcome::Done),
|
||||
"seal over a deleted tree must no-op to Done, got {outcome:?}"
|
||||
);
|
||||
queue_store::mark_done(&cfg, &claimed).unwrap();
|
||||
.expect("drain queue");
|
||||
assert_eq!(
|
||||
queue_store::get_job(&cfg, &job_id).unwrap().unwrap().status,
|
||||
queue_types::JobStatus::Done
|
||||
queue_types::JobStatus::Done,
|
||||
"seal over a deleted tree must settle to Done, not stuck pending"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -757,15 +752,16 @@ async fn queued_jobs_for_deleted_chunk_settle_to_done() {
|
||||
delete_chunks_by_source(&cfg, SourceKind::Chat, "slack:#eng").unwrap();
|
||||
assert!(get_chunk(&cfg, &c.id).unwrap().is_none());
|
||||
|
||||
queue_store::enqueue(
|
||||
let extract_id = queue_store::enqueue(
|
||||
&cfg,
|
||||
&queue_types::NewJob::extract_chunk(&queue_types::ExtractChunkPayload {
|
||||
chunk_id: c.id.clone(),
|
||||
})
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
queue_store::enqueue(
|
||||
.unwrap()
|
||||
.expect("extract_chunk job enqueued");
|
||||
let append_id = queue_store::enqueue(
|
||||
&cfg,
|
||||
&queue_types::NewJob::append_buffer(&queue_types::AppendBufferPayload {
|
||||
node: queue_types::NodeRef::Leaf {
|
||||
@@ -777,21 +773,21 @@ async fn queued_jobs_for_deleted_chunk_settle_to_done() {
|
||||
})
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
.unwrap()
|
||||
.expect("append_buffer job enqueued");
|
||||
|
||||
for _ in 0..2 {
|
||||
let job = queue_store::claim_next(&cfg, queue_store::DEFAULT_LOCK_DURATION_MS)
|
||||
.unwrap()
|
||||
.expect("job claimable");
|
||||
let outcome = crate::openhuman::memory_queue::handlers::handle_job(&cfg, &job)
|
||||
.await
|
||||
.expect("handle_job ok");
|
||||
assert!(
|
||||
matches!(outcome, queue_types::JobOutcome::Done),
|
||||
"{:?} over a deleted chunk must settle Done, got {outcome:?}",
|
||||
job.kind
|
||||
// Drive both queued jobs through the flipped crate engine (via the host
|
||||
// `drain_until_idle` → `worker::run_once` seam). Each must settle to Done
|
||||
// (warn-and-skip over the deleted chunk), not stay stuck pending.
|
||||
crate::openhuman::memory_queue::drain_until_idle(&cfg)
|
||||
.await
|
||||
.expect("drain queue");
|
||||
for id in [extract_id, append_id] {
|
||||
assert_eq!(
|
||||
queue_store::get_job(&cfg, &id).unwrap().unwrap().status,
|
||||
queue_types::JobStatus::Done,
|
||||
"job {id} over a deleted chunk must settle to Done"
|
||||
);
|
||||
queue_store::mark_done(&cfg, &job).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -31,7 +31,6 @@ use std::collections::{HashMap, HashSet};
|
||||
use anyhow::Result;
|
||||
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::memory_queue::handlers::chunk_tree_scope;
|
||||
use crate::openhuman::memory_store::chunks::store::{list_chunks, ListChunksQuery};
|
||||
use crate::openhuman::memory_store::chunks::types::{Chunk, SourceKind};
|
||||
use crate::openhuman::memory_store::content::read as content_read;
|
||||
@@ -40,6 +39,7 @@ use crate::openhuman::memory_tree::retrieval::types::{
|
||||
hit_from_chunk, hit_from_summary, QueryResponse, RetrievalHit,
|
||||
};
|
||||
use crate::openhuman::memory_tree::tree::store;
|
||||
use crate::openhuman::tinycortex::chunk_tree_scope;
|
||||
|
||||
/// Default cap on returned cover items when the caller passes `limit = 0`.
|
||||
const DEFAULT_LIMIT: usize = 200;
|
||||
|
||||
@@ -38,10 +38,18 @@ mod config;
|
||||
mod embeddings;
|
||||
#[cfg(test)]
|
||||
mod parity;
|
||||
mod queue_driver;
|
||||
|
||||
pub use chat::{build_chat_provider, SeamChatProvider};
|
||||
pub use config::memory_config_from;
|
||||
pub use embeddings::SeamEmbedder;
|
||||
pub use queue_driver::{
|
||||
classify_worker_error, HostQueueDelegates, WorkerErrorAction, WorkerReport,
|
||||
};
|
||||
// `chunk_tree_scope` is a crate-internal pure helper (the single canonical host
|
||||
// copy after the W4 handlers deletion); read paths like
|
||||
// `memory_tree::retrieval::cover` import it through this seam.
|
||||
pub(crate) use queue_driver::chunk_tree_scope;
|
||||
|
||||
// Facade re-exports — the rest of the host imports memory-engine types through
|
||||
// this one seam so consumer import paths stay stable as the internals flip to
|
||||
|
||||
@@ -88,6 +88,162 @@ mod tests {
|
||||
assert!(p1.ends_with(".md"), "chunk files are markdown -> {p1}");
|
||||
}
|
||||
|
||||
/// P6 (differential) — the host and crate `chunk_rel_path` must produce
|
||||
/// **byte-identical** vault paths for every id shape a real workspace holds.
|
||||
/// Both impls still exist (content is not flipped until W3), so a crate-side
|
||||
/// change to `slugify_source_id` / `sanitize_filename` / the email special
|
||||
/// case would silently strand every existing chunk file under a new path.
|
||||
/// This pins them together over an adversarial corpus (colons, all
|
||||
/// Windows-illegal chars, unicode, >255-char ids, gmail participant slugs,
|
||||
/// malformed email source_ids) so any drift fails here, not on a user's disk.
|
||||
#[test]
|
||||
fn chunk_rel_path_host_crate_byte_parity() {
|
||||
use crate::openhuman::memory_store::content::paths as host;
|
||||
use tinycortex::memory::store::content as cortex;
|
||||
|
||||
let long_id = "x".repeat(300);
|
||||
let corpus: &[(&str, &str, &str)] = &[
|
||||
// (source_kind, source_id, chunk_id)
|
||||
("chat", "slack:#eng", "chat:slack:#eng:0"),
|
||||
("chat", "Slack:#Eng__Team", "chat:slack:#eng:0"),
|
||||
("document", "file:///Users/x/Notes.md", "doc:notes:3"),
|
||||
("document", "weird__source__id", "id-with-no-illegal-chars"),
|
||||
("chat", "src", "a\\b/c:d*e?f\"g<h>i|j"),
|
||||
("chat", "东京:room", "chat:东京:0"),
|
||||
("chat", "src", &long_id),
|
||||
// Email: well-formed gmail participants → one slugified folder.
|
||||
(
|
||||
"email",
|
||||
"gmail:notifications@github.com|sanil@x.com",
|
||||
"email:msg:0",
|
||||
),
|
||||
("email", "gmail:Alice@X.com|bob@y.com", "email:msg:1"),
|
||||
// Email: malformed / legacy source_id → flat fallback layout.
|
||||
("email", "legacyid", "email:legacy:0"),
|
||||
("email", "gmail:", "email:empty-participants:0"),
|
||||
];
|
||||
|
||||
for (kind, source_id, chunk_id) in corpus {
|
||||
let h = host::chunk_rel_path(kind, source_id, chunk_id);
|
||||
let c = cortex::chunk_rel_path(kind, source_id, chunk_id);
|
||||
assert_eq!(
|
||||
h, c,
|
||||
"chunk_rel_path diverged for (kind={kind}, source_id={source_id}, chunk_id={chunk_id}): host={h} crate={c}"
|
||||
);
|
||||
assert!(!h.contains(':'), "host path leaked ':' -> {h}");
|
||||
assert!(h.ends_with(".md"), "chunk files are markdown -> {h}");
|
||||
}
|
||||
}
|
||||
|
||||
/// P6 (differential) — the same byte-parity requirement for summary paths.
|
||||
/// The summary basename (`summary_filename`) and the `wiki/summaries/...`
|
||||
/// layout per `SummaryTreeKind` must match across host and crate, or a
|
||||
/// re-open would not find an existing sealed summary in place.
|
||||
#[test]
|
||||
fn summary_rel_path_host_crate_byte_parity() {
|
||||
use crate::openhuman::memory_store::content::paths as host;
|
||||
use tinycortex::memory::store::content as cortex;
|
||||
|
||||
// (host kind, crate kind, scope_slug) — variants are 1:1 across sides.
|
||||
let kinds = [
|
||||
(
|
||||
host::SummaryTreeKind::Source,
|
||||
cortex::SummaryTreeKind::Source,
|
||||
"source-slug",
|
||||
),
|
||||
(
|
||||
host::SummaryTreeKind::Global,
|
||||
cortex::SummaryTreeKind::Global,
|
||||
"ignored-for-global",
|
||||
),
|
||||
(
|
||||
host::SummaryTreeKind::Topic,
|
||||
cortex::SummaryTreeKind::Topic,
|
||||
"phoenix-migration",
|
||||
),
|
||||
];
|
||||
// Canonical ms-first ids, legacy level-first ids, and malformed shapes
|
||||
// that must fall back through `sanitize_filename` on both sides.
|
||||
let summary_ids: &[&str] = &[
|
||||
"summary:1700000000000:L2-abc-uuid",
|
||||
"summary:L3:legacy-uuid",
|
||||
"summary:1700000000000:L2-a/b", // illegal tail → sanitized
|
||||
"summary:notms:L1-tail", // non-13-digit ms → fallback
|
||||
"raw-unknown-shape:with:colons", // unknown → sanitize_filename
|
||||
"东京-summary", // unicode
|
||||
];
|
||||
|
||||
for (hk, ck, scope) in kinds {
|
||||
for level in [0u32, 1, 4] {
|
||||
for sid in summary_ids {
|
||||
let h = host::summary_rel_path(hk, scope, level, sid);
|
||||
let c = cortex::summary_rel_path(ck, scope, level, sid);
|
||||
assert_eq!(
|
||||
h, c,
|
||||
"summary_rel_path diverged for (scope={scope}, level={level}, id={sid}): host={h} crate={c}"
|
||||
);
|
||||
assert!(!h.contains(':'), "host summary path leaked ':' -> {h}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// P10 — the embedding-space **signature** string that keys every persisted
|
||||
/// vector. Host (`embeddings::format_embedding_signature`) and crate
|
||||
/// (`store::vectors::format_embedding_signature`) each own their **own** copy
|
||||
/// of this formatter, so a change to either would silently split one
|
||||
/// embedding space into two — every existing vector would look stale under
|
||||
/// the new signature and trigger a full re-embed storm on the next open.
|
||||
/// Pin both to the golden `provider={name};model={model};dims={dims}` form
|
||||
/// over a corpus (real provider triples plus empties / special chars).
|
||||
#[test]
|
||||
fn embedding_signature_host_crate_byte_parity() {
|
||||
use crate::openhuman::embeddings::format_embedding_signature as host_sig;
|
||||
use tinycortex::memory::store::vectors::format_embedding_signature as cortex_sig;
|
||||
|
||||
// (name, model_id, dims, expected golden)
|
||||
let corpus: &[(&str, &str, usize, &str)] = &[
|
||||
(
|
||||
"voyage",
|
||||
"voyage-3",
|
||||
1024,
|
||||
"provider=voyage;model=voyage-3;dims=1024",
|
||||
),
|
||||
(
|
||||
"openai",
|
||||
"text-embedding-3-small",
|
||||
1536,
|
||||
"provider=openai;model=text-embedding-3-small;dims=1536",
|
||||
),
|
||||
(
|
||||
"ollama",
|
||||
"nomic-embed-text",
|
||||
768,
|
||||
"provider=ollama;model=nomic-embed-text;dims=768",
|
||||
),
|
||||
(
|
||||
"cohere",
|
||||
"embed-english-v3.0",
|
||||
1024,
|
||||
"provider=cohere;model=embed-english-v3.0;dims=1024",
|
||||
),
|
||||
("inert", "none", 0, "provider=inert;model=none;dims=0"),
|
||||
// Edge shapes: empty model, punctuation in model id.
|
||||
("noop", "", 3, "provider=noop;model=;dims=3"),
|
||||
("x", "m-1_2.3", 42, "provider=x;model=m-1_2.3;dims=42"),
|
||||
];
|
||||
|
||||
for (name, model, dims, golden) in corpus {
|
||||
let h = host_sig(name, model, *dims);
|
||||
let c = cortex_sig(name, model, *dims);
|
||||
assert_eq!(
|
||||
h, c,
|
||||
"signature diverged for (name={name}, model={model}, dims={dims}): host={h} crate={c}"
|
||||
);
|
||||
assert_eq!(&h, golden, "signature format drifted from the golden form");
|
||||
}
|
||||
}
|
||||
|
||||
fn hex(bytes: &[u8]) -> String {
|
||||
use std::fmt::Write;
|
||||
bytes
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,289 @@
|
||||
//! Layer-2 golden-workspace schema-parity harness (migration spec §0.3, parity
|
||||
//! checklist "Layer 2").
|
||||
//!
|
||||
//! The Layer-1 asserters (`src/openhuman/tinycortex/parity.rs`) pin pure on-disk
|
||||
//! *format* contracts (chunk ids, vector encoding, vault paths, signatures).
|
||||
//! This is the Layer-2 **differential** guard: it stands up a real workspace
|
||||
//! through the host's production memory surface (`memory::ops`) and asserts that
|
||||
//! the two schema tiers that share the workspace **compose** correctly —
|
||||
//!
|
||||
//! 1. the **crate-owned substrate** the `tinycortex` chunk DB creates
|
||||
//! (`init_db` → `chunks/schema.rs`), and
|
||||
//! 2. the **host-retained `UnifiedMemory` namespace-document tier**
|
||||
//! (`memory_store/unified/*`),
|
||||
//!
|
||||
//! coexisting without collision (parity checklist P3/P5/P11/P12 — the W3 gate).
|
||||
//! A store/tree cutover that reshaped, renamed, or dropped a table would strand
|
||||
//! an existing user workspace; this fails here first.
|
||||
//!
|
||||
//! Design notes:
|
||||
//! - **Path-agnostic.** It recursively scans *every* `*.db` under the temp
|
||||
//! workspace and unions their tables, so it does not care whether the tiers
|
||||
//! live in one DB file or several, nor exactly where the host client roots
|
||||
//! them.
|
||||
//! - The crate chunk-DB init is additionally forced via
|
||||
//! `tinycortex::memory::chunks::with_connection` so the substrate schema is
|
||||
//! deterministic regardless of which subsystems the op flow happened to touch.
|
||||
//! - `vectors` / `store_meta` / `kv_*` are created by other crate subsystems on
|
||||
//! their own first touch (the chunk/embed pipeline) rather than by the minimal
|
||||
//! doc-put + recall flow; they are reported in the run's schema dump and left
|
||||
//! to a follow-up that widens the flow, so this harness stays green and useful
|
||||
//! today without a fragile dependence on ingest internals.
|
||||
//!
|
||||
//! Run with: `cargo test --test memory_golden_parity_e2e`
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
|
||||
use tempfile::tempdir;
|
||||
|
||||
use openhuman_core::openhuman::config::Config;
|
||||
use openhuman_core::openhuman::memory::ops::{
|
||||
doc_put, memory_recall_context, memory_recall_memories, PutDocParams,
|
||||
};
|
||||
use openhuman_core::openhuman::memory::rpc_models::{RecallContextRequest, RecallMemoriesRequest};
|
||||
use openhuman_core::openhuman::tinycortex::memory_config_from;
|
||||
|
||||
// ── Env isolation (mirrors memory_roundtrip_e2e) ─────────────────────────────
|
||||
|
||||
struct EnvVarGuard {
|
||||
key: &'static str,
|
||||
old: Option<String>,
|
||||
}
|
||||
|
||||
impl EnvVarGuard {
|
||||
fn set_to_path(key: &'static str, path: &Path) -> Self {
|
||||
let old = std::env::var(key).ok();
|
||||
// SAFETY: only used under env_lock(), which serialises env mutation.
|
||||
unsafe { std::env::set_var(key, path.as_os_str()) };
|
||||
Self { key, old }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for EnvVarGuard {
|
||||
fn drop(&mut self) {
|
||||
match &self.old {
|
||||
// SAFETY: see set_to_path; teardown runs under the same env_lock().
|
||||
Some(v) => unsafe { std::env::set_var(self.key, v) },
|
||||
None => unsafe { std::env::remove_var(self.key) },
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Serialises tests: `HOME` + `OPENHUMAN_WORKSPACE` are process-global.
|
||||
static ENV_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
|
||||
|
||||
fn env_lock() -> std::sync::MutexGuard<'static, ()> {
|
||||
ENV_LOCK
|
||||
.get_or_init(|| Mutex::new(()))
|
||||
.lock()
|
||||
.expect("env lock poisoned")
|
||||
}
|
||||
|
||||
// ── Expected schema tiers (authoritative names from the two engines) ─────────
|
||||
|
||||
/// The crate chunk-DB substrate created by `init_db` (`chunks/schema.rs`). These
|
||||
/// are the tables the tinycortex store owns and must preserve byte-for-byte
|
||||
/// across every W3+ cutover.
|
||||
const CRATE_CHUNK_SCHEMA_TABLES: &[&str] = &[
|
||||
"mem_tree_chunks",
|
||||
"mem_tree_chunk_embeddings",
|
||||
"mem_tree_chunk_reembed_skipped",
|
||||
"mem_tree_score",
|
||||
"mem_tree_entity_index",
|
||||
"mem_tree_entity_edges",
|
||||
"mem_tree_trees",
|
||||
"mem_tree_summaries",
|
||||
"mem_tree_summary_embeddings",
|
||||
"mem_tree_summary_reembed_skipped",
|
||||
"mem_tree_buffers",
|
||||
"mem_tree_entity_hotness",
|
||||
"mem_tree_jobs",
|
||||
"mem_tree_ingested_sources",
|
||||
"mcp_writes",
|
||||
];
|
||||
|
||||
/// The host-retained `UnifiedMemory` namespace-document tier
|
||||
/// (`memory_store/unified/*`) — stays host, coexists in the shared workspace.
|
||||
const HOST_UNIFIED_TABLES: &[&str] = &[
|
||||
"memory_docs",
|
||||
"graph_global",
|
||||
"graph_namespace",
|
||||
"episodic_log",
|
||||
"event_log",
|
||||
"event_embeddings",
|
||||
"conversation_segments",
|
||||
"segment_embeddings",
|
||||
"vector_chunks",
|
||||
"user_profile",
|
||||
];
|
||||
|
||||
// ── Schema scan helpers (path-agnostic, read-only) ───────────────────────────
|
||||
|
||||
fn collect_db_files(dir: &Path, out: &mut Vec<PathBuf>) {
|
||||
let Ok(entries) = std::fs::read_dir(dir) else {
|
||||
return;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let path = entry.path();
|
||||
if path.is_dir() {
|
||||
collect_db_files(&path, out);
|
||||
} else if path.extension().and_then(|e| e.to_str()) == Some("db") {
|
||||
out.push(path);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Union of every user table across every `*.db` under `ws` (read-only opens;
|
||||
/// SQLite-internal `sqlite_%` tables excluded).
|
||||
fn tables_in_workspace(ws: &Path) -> BTreeSet<String> {
|
||||
let mut dbs = Vec::new();
|
||||
collect_db_files(ws, &mut dbs);
|
||||
|
||||
let mut tables = BTreeSet::new();
|
||||
for db in dbs {
|
||||
let Ok(conn) =
|
||||
rusqlite::Connection::open_with_flags(&db, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let Ok(mut stmt) = conn.prepare(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'",
|
||||
) else {
|
||||
continue;
|
||||
};
|
||||
let Ok(rows) = stmt.query_map([], |row| row.get::<_, String>(0)) else {
|
||||
continue;
|
||||
};
|
||||
for name in rows.flatten() {
|
||||
tables.insert(name);
|
||||
}
|
||||
}
|
||||
tables
|
||||
}
|
||||
|
||||
fn put_params(ns: &str) -> PutDocParams {
|
||||
PutDocParams {
|
||||
namespace: ns.to_string(),
|
||||
key: "golden-parity-canary".to_string(),
|
||||
title: "Golden parity canary".to_string(),
|
||||
content: "TinyCortex golden-workspace schema-parity canary fact".to_string(),
|
||||
source_type: "doc".to_string(),
|
||||
priority: "medium".to_string(),
|
||||
tags: Vec::new(),
|
||||
metadata: serde_json::Value::Null,
|
||||
category: "core".to_string(),
|
||||
session_id: None,
|
||||
document_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Drive the real production surface so both schema tiers initialise, then force
|
||||
/// the crate substrate init to make the chunk-DB schema deterministic. Returns
|
||||
/// the union of tables observed across the workspace.
|
||||
async fn init_and_scan(ns: &str, workspace: &Path) -> BTreeSet<String> {
|
||||
// Host unified tier + retrieval (production path).
|
||||
doc_put(put_params(ns)).await.expect("doc_put");
|
||||
let _ = memory_recall_memories(RecallMemoriesRequest {
|
||||
namespace: ns.to_string(),
|
||||
min_retention: None,
|
||||
as_of: None,
|
||||
limit: Some(10),
|
||||
max_chunks: None,
|
||||
top_k: None,
|
||||
})
|
||||
.await
|
||||
.expect("recall_memories");
|
||||
let _ = memory_recall_context(RecallContextRequest {
|
||||
namespace: ns.to_string(),
|
||||
include_references: Some(true),
|
||||
limit: Some(10),
|
||||
max_chunks: None,
|
||||
})
|
||||
.await
|
||||
.expect("recall_context");
|
||||
|
||||
// Force the crate chunk-DB substrate init (deterministic — creates the full
|
||||
// chunks/schema.rs table set regardless of what the ops above touched).
|
||||
let mc = memory_config_from(&Config::default(), workspace.to_path_buf());
|
||||
tinycortex::memory::chunks::with_connection(&mc, |_conn| Ok(())).expect("crate chunk-DB init");
|
||||
|
||||
tables_in_workspace(workspace)
|
||||
}
|
||||
|
||||
// ── Tests ────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// P3/P5/P11/P12 — the crate substrate and the host `UnifiedMemory` tier both
|
||||
/// initialise into the shared workspace without collision. Any cutover that
|
||||
/// renames/drops one of these tables fails here before it can strand a real
|
||||
/// user workspace.
|
||||
#[tokio::test]
|
||||
async fn golden_workspace_composes_substrate_and_unified_tiers() {
|
||||
let _lock = env_lock();
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let _home = EnvVarGuard::set_to_path("HOME", tmp.path());
|
||||
let workspace = tmp.path().join("workspace");
|
||||
std::fs::create_dir_all(&workspace).expect("mkdir workspace");
|
||||
let _ws = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", &workspace);
|
||||
|
||||
let tables = init_and_scan("golden-parity-e2e", &workspace).await;
|
||||
|
||||
// Full schema dump for review / manifest capture in the test log.
|
||||
eprintln!(
|
||||
"[golden-parity] workspace tables ({}): {:?}",
|
||||
tables.len(),
|
||||
tables
|
||||
);
|
||||
|
||||
let missing_substrate: Vec<&str> = CRATE_CHUNK_SCHEMA_TABLES
|
||||
.iter()
|
||||
.copied()
|
||||
.filter(|t| !tables.contains(*t))
|
||||
.collect();
|
||||
assert!(
|
||||
missing_substrate.is_empty(),
|
||||
"crate chunk-DB substrate tables missing from the workspace: {missing_substrate:?}; found: {tables:?}"
|
||||
);
|
||||
|
||||
let missing_unified: Vec<&str> = HOST_UNIFIED_TABLES
|
||||
.iter()
|
||||
.copied()
|
||||
.filter(|t| !tables.contains(*t))
|
||||
.collect();
|
||||
assert!(
|
||||
missing_unified.is_empty(),
|
||||
"host UnifiedMemory tables missing from the workspace: {missing_unified:?}; found: {tables:?}"
|
||||
);
|
||||
|
||||
// Coexistence: both tiers are present in the same workspace (P12).
|
||||
assert!(
|
||||
tables.contains("mem_tree_chunks") && tables.contains("memory_docs"),
|
||||
"both the crate substrate and the host unified tier must coexist"
|
||||
);
|
||||
}
|
||||
|
||||
/// Comparator 5 (idempotent re-open) — re-running the production flow over an
|
||||
/// existing workspace must not add or drop tables (no lazy second-run schema
|
||||
/// churn / migration storm).
|
||||
#[tokio::test]
|
||||
async fn golden_workspace_reopen_is_schema_stable() {
|
||||
let _lock = env_lock();
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let _home = EnvVarGuard::set_to_path("HOME", tmp.path());
|
||||
let workspace = tmp.path().join("workspace");
|
||||
std::fs::create_dir_all(&workspace).expect("mkdir workspace");
|
||||
let _ws = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", &workspace);
|
||||
|
||||
let first = init_and_scan("golden-parity-reopen", &workspace).await;
|
||||
let second = init_and_scan("golden-parity-reopen", &workspace).await;
|
||||
|
||||
assert_eq!(
|
||||
first, second,
|
||||
"re-running the flow changed the workspace table set (schema churn on re-open)"
|
||||
);
|
||||
assert!(
|
||||
!first.is_empty(),
|
||||
"sanity: the first pass should have created tables"
|
||||
);
|
||||
}
|
||||
Vendored
+1
-1
Submodule vendor/tinycortex updated: 33dda94305...a8e10f7dd8
Reference in New Issue
Block a user