From ed4d28cb5ade337030775fa727893e11623a816c Mon Sep 17 00:00:00 2001 From: eunhwa99 Date: Mon, 15 Jun 2026 17:22:44 +0900 Subject: [PATCH 1/2] fix: harden background sync_source lifecycle --- ...-sync-source-background-launch-contract.md | 97 +++ .agents/docs/adr/README.md | 1 + .agents/docs/architecture.md | 5 +- README.md | 6 +- api/tools.py | 9 +- docs/contextwiki-core-understanding.md | 43 +- .../2026-06-15-notion-cancel-sync-stuck.md | 183 ++++++ indexing/ingestion_service.py | 163 ++++- scripts/demo_public_flow.py | 18 +- tests/api/test_tools_contract.py | 33 +- tests/contracts/test_public_mcp_contracts.py | 126 +++- tests/e2e/test_contextwiki_flow.py | 135 +++- tests/e2e/test_obsidian_connector_flow.py | 30 +- tests/e2e/test_phase_b_connectors_flow.py | 48 +- tests/indexing/test_ingestion_service.py | 613 ++++++++++++++++++ tests/scripts/test_demo_public_flow.py | 26 +- 16 files changed, 1444 insertions(+), 92 deletions(-) create mode 100644 .agents/docs/adr/0007-sync-source-background-launch-contract.md create mode 100644 docs/plan/2026-06-15-notion-cancel-sync-stuck.md diff --git a/.agents/docs/adr/0007-sync-source-background-launch-contract.md b/.agents/docs/adr/0007-sync-source-background-launch-contract.md new file mode 100644 index 0000000..2b77a9f --- /dev/null +++ b/.agents/docs/adr/0007-sync-source-background-launch-contract.md @@ -0,0 +1,97 @@ +# ADR 0007: Public `sync_source` Background Launch Contract + +## Status + +accepted + +## Date + +2026-06-15 + +## Context + +`MCPContentSearch` originally treated `sync_source(source_id)` as a blocking +end-to-end sync call. That worked for short source fetches, but it breaks down +for larger configured sources such as Notion when the MCP client, transport, or +caller timeout cancels the request before the sync finishes. In the observed +Notion failure, the configured `NOTION_API_KEY` was valid and live requests were +still returning `200 OK`, but the client disconnected during a long sync and +left the SQLite job status in `running`. + +The repo already uses SQLite metadata as the authoritative source/job state +store under ADR 0002, and existing MCP clients already depend on the retained +`sync_source` and `get_sync_status` tool names from ADR 0004/ADR 0006 scope. +The fix therefore needs to preserve the current MCP surface while changing the +public completion semantics so long-running syncs survive request lifetime. + +## Decision + +Keep two distinct sync boundaries: + +- Internal direct callers use `IngestionService.sync_source(source_id)` as the + blocking execution path when they start a new sync themselves or when they + can join a same-process local background task for that source. If another + owner already holds the running SQLite job and there is no joinable local + task, the direct path returns that current running job instead of pretending + it can block on foreign in-flight work. +- Public MCP callers use `IngestionService.start_sync_source(source_id)` through + the `sync_source` tool as an immediate-return background launcher. + +Under this contract, public MCP `sync_source(source_id)`: + +- starts a new background sync job or reuses the active running job for the + same source +- returns the current job payload immediately, typically with `status=running` +- requires callers to poll `get_sync_status(source_id)` for terminal + `succeeded` or `failed` completion +- must not silently fall back to the blocking path when a background launcher + is unavailable + +Background execution still owns the full fetch, chunk, index, metadata commit, +and stale-cleanup lifecycle. Cancellation or early background failure must +reconcile the job out of `running` through SQLite metadata so later retries are +not blocked by stuck state. When a direct same-process caller encounters a +just-cancelled local background task, it may surface that reconciled failed job +once instead of opening fresh duplicate work immediately, but only while that +same cancelled job is still the latest authoritative SQLite job for the +source. This exception is for cancelled-local handoff only; a successfully +completed background sync must not suppress the next fresh direct sync, and a +newer foreign retry must override the stale local cancelled handoff. + +## Consequences + +- MCP tool names remain stable: callers keep using `sync_source` and + `get_sync_status` instead of introducing a new public launcher tool. +- Client-facing docs, architecture notes, and retained MCP contract tests must + describe and verify the immediate-return plus polling behavior. +- Review gates must treat public `sync_source` as a contract boundary change, + not just an implementation detail. +- Background-task cancellation, early startup failure, and overlapping + same-source launch reuse need focused regression coverage because they can + otherwise leave stale `running` jobs behind. +- Direct service-level tests and internal call sites can continue to rely on the + blocking `IngestionService.sync_source()` path when they start the work + themselves or join a same-process local background task. They must not assume + cross-process blocking over a foreign running job that only exists in SQLite + metadata. + +## Alternatives Considered + +- Keep public `sync_source` blocking and only raise client timeouts: rejected + because the real failure mode was a cancelled long-running sync leaving + incorrect SQLite state. +- Add a brand-new MCP tool such as `start_sync_source`: rejected because the + current retained MCP surface can support the new behavior without expanding + the public tool set. +- Convert every sync caller to background-only semantics, including direct + service callers: rejected because internal focused tests and direct service + flows still benefit from a blocking path with immediate terminal state. + +## Related + +- `.agents/docs/architecture.md` +- `.agents/docs/adr/0002-contextwiki-metadata-and-citation-store.md` +- `.agents/docs/adr/0004-contextwiki-phase-b-connectors.md` +- `.agents/docs/adr/0006-slim-mcp-core-scope.md` +- `docs/contextwiki-core-understanding.md` +- `docs/plan/2026-06-15-notion-cancel-sync-stuck.md` diff --git a/.agents/docs/adr/README.md b/.agents/docs/adr/README.md index c31220b..04f6832 100644 --- a/.agents/docs/adr/README.md +++ b/.agents/docs/adr/README.md @@ -32,6 +32,7 @@ File names should be numbered and descriptive: | [0004](0004-contextwiki-phase-b-connectors.md) | accepted, website/docs superseded by [0006](0006-slim-mcp-core-scope.md) | ContextWiki Phase B GitHub connector, retained Obsidian local-vault connector, and superseded website/docs connector | | [0005](0005-contextwiki-auto-wiki-llm-synthesis.md) | superseded by [0006](0006-slim-mcp-core-scope.md) for current scope | Historical ContextWiki Auto Wiki LLM synthesis boundary | | [0006](0006-slim-mcp-core-scope.md) | accepted | Slim MCP core scope | +| [0007](0007-sync-source-background-launch-contract.md) | accepted | Public MCP `sync_source` background-launch contract and internal blocking execution split | ## When to Add or Update ADRs diff --git a/.agents/docs/architecture.md b/.agents/docs/architecture.md index 4b8d95d..54efc57 100644 --- a/.agents/docs/architecture.md +++ b/.agents/docs/architecture.md @@ -60,13 +60,16 @@ Source sync flow: ```text sync_source - -> IngestionService + -> IngestionService.start_sync_source() for MCP callers -> SourceRegistry connector lookup -> MetadataStore source registration and sync job guard + -> immediate running-job payload returned to caller + -> background IngestionService worker fetch/index lifecycle -> Notion, Tistory, GitHub, or Obsidian connector fetch -> DocumentChunker -> ContentIndexer and Chroma collection -> MetadataStore SQLite source/job/document/chunk/tombstone metadata + -> get_sync_status reads terminal completion ``` Retrieval and answer flow: diff --git a/README.md b/README.md index 558b242..8f38b96 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,7 @@ Tool handlers live in `api/tools.py`. Business logic stays in `fetching/`, | Tool | Purpose | | --- | --- | | `list_sources()` | List configured Notion, Tistory, GitHub, and Obsidian sources. | -| `sync_source(source_id)` | Sync one configured source into SQLite metadata and Chroma vectors. | +| `sync_source(source_id)` | Start or reuse one configured source sync job and return the current job state immediately. Poll `get_sync_status(source_id)` for terminal completion. | | `sync_all()` | Sync all retained sources concurrently and return aggregate results. | | `get_sync_status(source_id="")` | Read latest source and sync-job state. | | `search_context(query, filters=None, top_k=10, include_debug=False)` | Return structured evidence chunks after Chroma retrieval, metadata fallback when needed, and SQLite validation. | @@ -57,6 +57,10 @@ Tool handlers live in `api/tools.py`. Business logic stays in `fetching/`, At a glance: - `sync_all()` syncs all retained sources in one pass. +- `sync_source(source_id)` is an immediate-return launcher for MCP clients. It + starts or reuses the current sync job for that source, typically returning a + `running` job payload immediately, and long sync completion should be + observed through `get_sync_status(source_id)`. - `sync_all()` reports aggregate `status` truthfully: `completed` for succeed/skip-only runs, `partial` when success/skip is mixed with blocked or failed sources, and `failed` when nothing completed successfully. diff --git a/api/tools.py b/api/tools.py index e33e009..90355c2 100644 --- a/api/tools.py +++ b/api/tools.py @@ -76,11 +76,16 @@ async def list_sources() -> dict: @mcp.tool() async def sync_source(source_id: str) -> dict: - """특정 source incremental sync 실행""" + """특정 source sync를 시작하거나 기존 running job을 재사용한다.""" if ingestion_service is None: return {"status": "error", "message": "ingestion service is not configured"} + if not hasattr(ingestion_service, "start_sync_source"): + return { + "status": "error", + "message": "ingestion service does not support background sync launch", + } try: - job = await ingestion_service.sync_source(source_id) + job = await ingestion_service.start_sync_source(source_id) return _safe_sync_job_payload(job) except Exception as exc: message = safe_error_message(exc) diff --git a/docs/contextwiki-core-understanding.md b/docs/contextwiki-core-understanding.md index 22700fb..ea3c216 100644 --- a/docs/contextwiki-core-understanding.md +++ b/docs/contextwiki-core-understanding.md @@ -110,10 +110,16 @@ Bulk retained-source sync is also available: sync_all() ``` -This fans out one concurrent `sync_source()` run per retained source. Each +This fans out one concurrent internal `sync_source()` run per retained source. Each source still keeps its own SQLite running-job guard, so a source that is already syncing is reported as blocked in the aggregate result instead of -starting a second overlapping fetch. +starting a second overlapping fetch. For the public MCP tool, `sync_source` +now starts or reuses the running job quickly and returns the initial `running` +job payload while the actual ingestion continues in a server-owned background +task. Callers should use `get_sync_status` to observe terminal `succeeded` or +`failed` completion. If the internal blocking ingestion path is cancelled before +completion, the ingestion layer marks that source job failed instead of leaving +a permanent `running` job behind, so a later retry can start cleanly. Obsidian example: @@ -190,7 +196,7 @@ Current tools: | Tool | Use | | --- | --- | | `list_sources()` | see configured sources | -| `sync_source(source_id)` | refresh one source | +| `sync_source(source_id)` | start one source sync quickly and return the current job | | `sync_all()` | refresh all retained sources concurrently and return aggregate results | | `get_sync_status(source_id?)` | inspect source/job state | | `search_context(query, filters, top_k, include_debug)` | find SQLite-validated evidence | @@ -203,6 +209,12 @@ On the normal default path `api/tools.py` returns `debug={}` unless `include_debug=True`, and when caller filters leave no matching public sources it still returns a small populated `debug` object even if `include_debug=False`. +The MCP `sync_source(source_id)` tool name and arguments are unchanged, but the +public tool now routes through `IngestionService.start_sync_source()`. A fresh +tool call therefore returns the initial `running` job promptly, or the existing +running job if one is already active. Callers should use +`get_sync_status(source_id)` to observe terminal completion. + The current debug payload includes rewrite decision fields: ```text @@ -440,8 +452,19 @@ indexing/indexer.py storage/metadata_store.py ``` -`IngestionService.sync_source()` is the core per-source business flow, and -`IngestionService.sync_all()` is the retained-source aggregate fan-out entrypoint. +`IngestionService.sync_source()` remains the blocking internal per-source +business flow when it starts the sync itself or joins a same-process local +background task. If another owner already holds the running SQLite job and +there is no local task to join, it returns that current `running` job instead +of blocking on foreign in-flight work. `IngestionService.start_sync_source()` +is the immediate-return background launcher used by the public MCP tool, and +`IngestionService.sync_all()` is the retained-source aggregate fan-out +entrypoint. A just-cancelled same-process background task may be surfaced once +as its reconciled failed job to a direct caller instead of immediately opening +duplicate work, but only while that cancelled job is still the latest +authoritative SQLite job for the source. A successfully completed background +sync does not suppress the next fresh direct `sync_source()` run, and a newer +foreign retry must override any stale local cancelled handoff. ```text sync_source(source_id) @@ -459,6 +482,16 @@ sync_source(source_id) -> best-effort vector cleanup ``` +```text +start_sync_source(source_id) +-> SourceRegistry connector lookup +-> MetadataStore register_source + begin_sync_job guard +-> if a job is already running, return that running job +-> otherwise spawn the blocking sync_source() work in a background task +-> return the initial running job immediately +-> caller polls get_sync_status(source_id) for succeeded / failed completion +``` + ```text sync_all() -> SourceRegistry retained source enumeration diff --git a/docs/plan/2026-06-15-notion-cancel-sync-stuck.md b/docs/plan/2026-06-15-notion-cancel-sync-stuck.md new file mode 100644 index 0000000..1e07c53 --- /dev/null +++ b/docs/plan/2026-06-15-notion-cancel-sync-stuck.md @@ -0,0 +1,183 @@ +# Notion Sync Cancellation Leaves Running Job Stuck + +## User Request + +Investigate why Notion does not sync properly and verify whether the configured +API key is failing. After confirming the key is not the problem, fix the root +cause for the stuck Notion sync. Follow-up: make long source syncs continue +even when the client times out, using the approved `sync_source` background-job +direction with explicit subagent/delegation approval. + +## Branch Preflight Result + +- Original worktree `/Users/eunhwa/IdeaProjects/MCPContentSearch` is dirty on + `main` (`.env.example` modified), so it was preserved untouched. +- Fetched `origin/main`. +- Created isolated worktree `/private/tmp/MCPContentSearch-notion-cancel-sync` + on fresh branch `feature/fix-notion-cancel-sync-stuck` from `origin/main`. +- Initial atomic cancellation-fix work was completed directly before the scope + expanded. +- User later approved a broader `sync_source` background-job change plus + explicit subagent/delegation use, so the remainder of this work item now uses + worker orchestration. + +## Scope And Non-goals + +Scope: + +- Confirm whether Notion auth is actually failing in the live Docker path. +- Fix the sync lifecycle so a cancelled MCP request does not leave a permanent + `running` job for `source_notion`. +- Add focused regression coverage around cancellation cleanup at the ingestion + service boundary. +- Keep `IngestionService.sync_source()` as the blocking internal execution path, + add `IngestionService.start_sync_source()` as the immediate-return background + launcher, and route the public MCP `sync_source` tool through that launcher. +- Add or update focused contract/E2E coverage for the new `running -> poll for + completion` behavior. + +Non-goals: + +- Do not change Notion fetch semantics, block parsing, or credentials policy. +- Do not inspect or mutate user Chroma content beyond the existing live debug + evidence already gathered. +- Do not change MCP tool shapes unless required for truthful lifecycle cleanup. + +## Acceptance Criteria + +- Live-debug evidence remains consistent with `NOTION_API_KEY` being loaded and + accepted by Notion, not rejected for auth. +- If `sync_source()` is cancelled during a long fetch, the sync job is no + longer left in `running` state forever. +- A subsequent sync attempt can start again after cancellation cleanup. +- Focused regression tests cover the cancellation path without live Notion or + user data. +- The public MCP `sync_source()` tool returns promptly with a truthful running + job when it starts a long sync. +- Direct internal callers of `IngestionService.sync_source()` still receive the + blocking completion path. +- The background worker continues after the request returns and eventually marks + the job/source succeeded or failed in SQLite metadata. +- A second public `sync_source()` tool call during active background work + reuses the same running job instead of starting duplicate work. + +## Step Breakdown + +1. Trace the live failure path from MCP request cancellation through + `IngestionService.sync_source()` and SQLite job ownership/status handling. +2. Keep the cancellation cleanup fix in place. +3. Keep direct service `sync_source()` blocking, add `start_sync_source()` for + the request-facing launcher, and let the background worker perform the long + ingestion lifecycle. +4. Update focused ingestion coverage for the new launcher and adapt MCP + contract/retained E2E coverage to the public `running -> poll` behavior. +5. Run broader verification and functional smoke entries relevant to source + sync, then proceed to delegated review. + +## Files Likely To Change + +- `indexing/ingestion_service.py` +- `api/tools.py` +- `tests/indexing/test_ingestion_service.py` +- `tests/api/test_tools_contract.py` +- `tests/e2e/test_contextwiki_flow.py` +- `tests/e2e/test_phase_b_connectors_flow.py` +- `docs/contextwiki-core-understanding.md` + +## Test And Verification Plan + +- Focused regression: `uv run pytest tests/indexing/test_ingestion_service.py -q` +- Focused contract regression: `uv run pytest tests/api/test_tools_contract.py -q` +- Syntax safety: `python -m compileall api core environments fetching indexing search storage main.py` +- Functional gate: `./scripts/verify_functional_e2e.sh` +- Retained MCP E2E regression: `uv run pytest tests/e2e/test_contextwiki_flow.py tests/e2e/test_phase_b_connectors_flow.py tests/e2e/test_obsidian_connector_flow.py -q` + +## Functional Smoke Matrix + +| Feature | Surface | Data Mode | Expected Result | Status | +| --- | --- | --- | --- | --- | +| Configured Notion sync cancellation cleanup | direct `IngestionService.sync_source` / metadata job state | fake/temp | Cancelled blocking sync ends in truthful non-running state and does not block later syncs | completed | +| Configured source retry after cancellation | direct `IngestionService.sync_source` / metadata job state | fake/temp | A new blocking sync can start after the cancelled job is reconciled | completed | +| Background configured sync launch | MCP `sync_source` / metadata job state | fake/temp | MCP `sync_source` returns a running job quickly while background work keeps going | completed | +| Background sync completion polling | MCP `get_sync_status` / retained E2E caller | fake/temp | Polling eventually observes `succeeded` or `failed` after background completion | completed | +| Live Docker Notion auth evidence | Docker logs / current metadata DB | live read-only | Notion requests continue to return `200 OK`, confirming auth is not the failure | completed | + +## Architecture And ADR Constraints + +- ADR 0002: SQLite metadata is the authoritative sync/job status store; cleanup + must keep source/job state truthful without storing raw secrets. +- ADR 0004: Connector-specific behavior stays inside existing fetching/indexing + boundaries; do not add connector-specific MCP tools for this fix. +- Architecture doc: keep API/tool contracts thin and put lifecycle logic in the + service/storage layers. + +## Risks And Rollback Notes + +- `asyncio` cancellation can bypass generic `except Exception` handling, so the + fix must explicitly handle cancellation without swallowing unrelated errors. +- Background task failures must be persisted, not dropped silently. +- Request-return semantics change from "usually completed job" toward "running + job launcher", so tests and documentation must be updated carefully. +- Cleanup must avoid falsely marking a still-active competing job as failed. +- Rollback is limited to reverting the cancellation-specific lifecycle changes + and their focused test coverage. + +## Progress Log + +| Phase | Status | Summary | Evidence | +| --- | --- | --- | --- | +| Branch preflight | completed | Preserved dirty `main`, fetched `origin/main`, and created isolated feature worktree. | `git status --short --branch`; `git fetch origin main`; `git worktree add -b feature/fix-notion-cancel-sync-stuck /private/tmp/MCPContentSearch-notion-cancel-sync origin/main` | +| Root-cause investigation | completed | Verified live Docker Notion requests return `200 OK`; stuck state comes from cancelled sync request leaving job `2b55e869-...` in `running` with unchanged heartbeat while fetch was still in progress. | Docker logs show `Found 265 Notion pages`, `Progress: 10/265`, then `Request cancelled`; SQLite `sync_jobs` row stays `running` with `heartbeat_at=2026-06-15T06:06:42.327570+00:00` | +| Planning | completed | Created this plan before target edits. | `docs/plan/2026-06-15-notion-cancel-sync-stuck.md` | +| Implementation | completed | Production ownership moved to the implementation worker: direct `IngestionService.sync_source()` stayed blocking, public MCP `sync_source` now routes through `start_sync_source()`, and this follow-up stayed scoped to owned tests/docs. | `indexing/ingestion_service.py`; `api/tools.py`; owned test/docs files only edited here | +| Boundary revision | completed | Adjusted the test/doc plan to the reduced-blast-radius boundary: direct service tests stay mostly unchanged, MCP-level tests poll `get_sync_status`. | User integration update on 2026-06-15; owned file scope confirmed | +| Focused verification | completed | Focused ingestion, API contract, public contract, and retained MCP E2E coverage passed with the revised boundary. | `uv run pytest tests/indexing/test_ingestion_service.py -q -k "start_sync_source_returns_running_job_and_completes_in_background or cancelled_source_sync_marks_job_failed_and_allows_retry"` -> 2 passed; `uv run pytest tests/api/test_tools_contract.py -q -k "sync_source_"` -> 4 passed; `uv run pytest tests/contracts/test_public_mcp_contracts.py -q -k "sync_source_contract or get_sync_status_contract"` -> 2 passed; `uv run pytest tests/e2e/test_contextwiki_flow.py -q -k "contextwiki_fake_e2e_sync_search_fetch_and_answer or contextwiki_temp_chroma_e2e_sync_search_fetch_and_answer or phase1_alias_expansion or phase2_query_rewrite or phase3_repository_lookup"` -> 5 passed; `uv run pytest tests/e2e/test_phase_b_connectors_flow.py -q -k "retained_source_smoke or retained_github_sync_through_mcp_tools"` -> 1 passed | +| Functional smoke | completed | Repo functional smoke passed after the retained E2E suite was updated for immediate-return `sync_source` plus `get_sync_status` polling. | `./scripts/verify_functional_e2e.sh` -> 25 passed in 4.90s | +| Scope expansion | completed | User approved the broader background-job direction for `sync_source` plus explicit delegation. | Approved `1+4`, then confirmed to proceed | +| Worker orchestration | completed | The follow-up initially stayed atomic within the owned test/doc scope after the boundary revision, and later review-driven fixes extended back into the owned production boundary without spawning a new split. | Ownership stayed inside the planned sync-source boundary across code, tests, docs, and plan updates | +| Verification refresh | completed | Re-ran syntax, focused regressions, retained MCP E2E coverage, full functional smoke, and diff hygiene in the isolated worktree before review. | `python -m compileall api core environments fetching indexing search storage main.py` passed; `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py -q` -> 86 passed in 2.97s; `uv run pytest tests/e2e/test_contextwiki_flow.py tests/e2e/test_phase_b_connectors_flow.py tests/e2e/test_obsidian_connector_flow.py -q` -> 25 passed in 5.07s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 4.90s; `git diff --check` passed | +| Review pass 1 | completed | Five fresh reviewers found one real race window around background-task cancellation plus public contract doc and MCP re-entry test gaps. | Findings referenced `indexing/ingestion_service.py`, `README.md`, `.agents/docs/architecture.md`, `api/tools.py`, and `tests/contracts/test_public_mcp_contracts.py` | +| Review fixes | completed | Hardened background-task cancellation reconciliation, added focused launcher reuse/cancellation regressions, expanded MCP contract coverage for same-job reuse plus polling, and updated public contract docs. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py`; `tests/contracts/test_public_mcp_contracts.py`; `README.md`; `.agents/docs/architecture.md`; `api/tools.py` | +| Review-fix verification | completed | Re-ran syntax, focused regression suites, retained MCP E2E suites, functional smoke, and diff hygiene after the review-driven fixes. | `python -m compileall api core environments fetching indexing search storage main.py` passed; `uv run pytest tests/indexing/test_ingestion_service.py tests/contracts/test_public_mcp_contracts.py tests/api/test_tools_contract.py -q` -> 89 passed in 3.31s; `uv run pytest tests/e2e/test_contextwiki_flow.py tests/e2e/test_phase_b_connectors_flow.py tests/e2e/test_obsidian_connector_flow.py -q` -> 25 passed in 6.01s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 5.52s; `git diff --check` passed | +| Review pass 2 | completed | Five fresh reviewers found one more contract-hardening gap in the MCP tool fallback path and one bounded-wait gap in retained E2E polling. | Findings referenced `api/tools.py`, `tests/e2e/test_contextwiki_flow.py`, and a small plan audit-trail wording mismatch | +| Review pass 2 fixes | completed | Removed the public MCP fallback to blocking `sync_source`, switched the retained FastMCP polling tests to bounded helper usage, and aligned the plan audit-trail wording. | `api/tools.py`; `tests/e2e/test_contextwiki_flow.py`; `docs/plan/2026-06-15-notion-cancel-sync-stuck.md` | +| Review pass 2 verification | completed | Re-ran the affected API/contract/retained E2E suites plus the repo functional smoke after the second-pass fixes. | `uv run pytest tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 50 passed in 5.61s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 5.55s; `git diff --check` passed | +| Review pass 3 | completed | Five fresh reviewers found one remaining early-failure window before the background worker entered `_run_sync_source_job()`'s guarded path. | Finding referenced `indexing/ingestion_service.py` initial `get_sync_job()` lookup before the main `try` block | +| Review pass 3 fixes | completed | Moved the initial metadata lookup under exception handling and added a regression that forces the first background `get_sync_job()` read to miss so the job is marked failed instead of staying `running`. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py` | +| Review pass 3 verification | completed | Re-ran syntax, focused ingestion/API/contract/retained E2E coverage, and repo functional smoke after the early-failure fix. | `python -m compileall api core environments fetching indexing search storage main.py` passed; `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 96 passed in 10.44s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 7.72s; `git diff --check` passed | +| Review pass 4 | completed | Five fresh reviewers found one remaining architecture-decision gap: the public MCP `sync_source` contract shift needed an ADR. | Finding referenced `.agents/docs/adr/README.md` policy and the new public `sync_source` polling contract | +| Review pass 4 fixes | completed | Added ADR 0007 to record the internal blocking vs public background-launch sync boundary, polling expectations, and cancellation ownership. | `.agents/docs/adr/0007-sync-source-background-launch-contract.md`; `.agents/docs/adr/README.md` | +| Review pass 4 verification | completed | Re-checked diff hygiene after the ADR/documentation-only follow-up. | `git diff --check` passed | +| Review pass 5 | completed | Five fresh reviewers found one remaining contract mismatch: a direct internal `sync_source()` call did not wait for a locally launched background sync on the same source. | Finding referenced `indexing/ingestion_service.py`, ADR 0007, and `docs/contextwiki-core-understanding.md` | +| Review pass 5 fixes | completed | Updated the blocking direct path to await the local in-flight background task for the same source and added a regression that launches background sync then joins it through direct `sync_source()`. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py` | +| Review pass 5 verification | completed | Re-ran focused ingestion/API/contract/retained E2E coverage and repo functional smoke after restoring direct blocking semantics. | `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 97 passed in 10.07s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 8.03s; `git diff --check` passed | +| Review pass 6 | completed | A fresh reviewer found one more aggregate-contract regression: `sync_all()` was joining a local background source sync instead of reporting it as `blocked`. | Finding referenced `indexing/ingestion_service.py`, `README.md`, and `docs/contextwiki-core-understanding.md` blocked semantics | +| Review pass 6 fixes | completed | Split the direct-join behavior from the bulk path so `sync_all()` keeps non-joining blocked semantics while direct `sync_source()` still joins a local background sync, and added a regression for `start_sync_source()` followed by `sync_all()`. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py` | +| Review pass 6 verification | completed | Re-ran focused ingestion/API/contract/retained E2E coverage and repo functional smoke after restoring `sync_all()` blocked behavior. | `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 98 passed in 10.06s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 8.17s; `git diff --check` passed | +| Review pass 7 | completed | A fresh reviewer found one remaining test-quality gap: the public-contract fake could recreate a new running job while still appearing to "reuse" because it hid `status`/`job_id` attributes and reused the same hard-coded id. | Finding referenced `tests/contracts/test_public_mcp_contracts.py` fake metadata/job behavior | +| Review pass 7 fixes | completed | Exposed fake job attributes in `Dumpable` storage and made the fake ingestion service issue distinct job ids so contract reuse tests fail if reuse breaks. | `tests/contracts/test_public_mcp_contracts.py` | +| Review pass 7 verification | completed | Re-ran the public MCP contract suite and diff hygiene after tightening the fake reuse semantics. | `uv run pytest tests/contracts/test_public_mcp_contracts.py -q` -> 10 passed in 0.47s; `git diff --check` passed | +| Review pass 8 | completed | A fresh reviewer found two final follow-ups outside the main sync engine: the public demo script still assumed blocking MCP sync, and an untracked design spec artifact was still present in the branch. | Findings referenced `scripts/demo_public_flow.py`, `tests/scripts/test_demo_public_flow.py`, and `docs/superpowers/specs/2026-06-15-background-sync-source-design.md` | +| Review pass 8 fixes | completed | Updated the public demo harness to poll `get_sync_status()` to a terminal job before retrieval/answer steps and removed the stray untracked design artifact from `docs/superpowers/specs/`. | `scripts/demo_public_flow.py`; deleted `docs/superpowers/specs/2026-06-15-background-sync-source-design.md` | +| Review pass 8 verification | completed | Re-ran the public demo-script suite and diff hygiene after the demo/artifact follow-up. | `uv run pytest tests/scripts/test_demo_public_flow.py -q` -> 16 passed in 25.28s; `git diff --check` passed | +| Review pass 9 | completed | Fresh reviewers found two last truthfulness gaps: joined-background pre-start cancellation still had a one-tick direct-caller race, and the public demo output was overwriting the launcher payload with the terminal job. | Findings referenced `indexing/ingestion_service.py`, `tests/indexing/test_ingestion_service.py`, `scripts/demo_public_flow.py`, and `tests/scripts/test_demo_public_flow.py` | +| Review pass 9 fixes | completed | Added reconcile-and-yield handling for joined-background pre-start cancellation, extended the regression coverage, and kept the demo's `sync` payload as the initial launcher response while leaving terminal completion in `status.latest_job`. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py`; `scripts/demo_public_flow.py`; `tests/scripts/test_demo_public_flow.py` | +| Review pass 9 verification | completed | Re-ran focused ingestion/API/contract/retained E2E coverage, the public demo-script suite, and the repo functional smoke after the final truthfulness fixes. | `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 101 passed in 13.85s; `uv run pytest tests/scripts/test_demo_public_flow.py -q` -> 16 passed in 27.72s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 10.47s; `git diff --check` passed | +| Review pass 10 | completed | A fresh reviewer flagged one remaining wording mismatch: the ADR/context note overstated blocking behavior for direct callers that observe a foreign running SQLite job with no joinable local task. | Finding referenced ADR 0007 and `docs/contextwiki-core-understanding.md` direct-caller wording | +| Review pass 10 fixes | completed | Narrowed the direct-caller wording so blocking semantics apply when the caller starts the work itself or joins a same-process local background task, while foreign running jobs are described as current-state returns. | `.agents/docs/adr/0007-sync-source-background-launch-contract.md`; `docs/contextwiki-core-understanding.md` | +| Review pass 10 verification | completed | Re-checked diff hygiene after the wording-only follow-up. | `git diff --check` passed | +| Review pass 11 | completed | A fresh reviewer found one last direct-caller race: after a local background task is cancelled and reconciled, `sync_source()` still needed to surface that terminal result once instead of immediately opening a fresh job. | Findings referenced `indexing/ingestion_service.py` pre-start cancel path and the matching regression in `tests/indexing/test_ingestion_service.py` | +| Review pass 11 fixes | completed | Stored the most recent reconciled terminal background job per source so direct `sync_source()` can return that one-shot terminal result, while `start_sync_source()` still clears it and opens a fresh retry. | `indexing/ingestion_service.py` | +| Review pass 11 verification | completed | Re-ran focused ingestion/API/contract/retained E2E coverage, the public demo-script suite, and the repo functional smoke after the direct-caller terminal-result fix. | `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 101 passed in 17.14s; `uv run pytest tests/scripts/test_demo_public_flow.py -q` -> 16 passed in 33.83s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 11.50s; `git diff --check` passed | +| Review pass 12 | completed | A fresh reviewer found one remaining cache-scope regression: the one-shot terminal background cache also captured successful background completions and could suppress the next fresh direct sync. | Finding referenced `indexing/ingestion_service.py` `_recent_terminal_background_jobs` scope and the missing post-success rerun regression | +| Review pass 12 fixes | completed | Restricted the one-shot terminal cache to cancelled-to-failed background reconciliations only and added a regression that a successful background completion is followed by a fresh direct sync with a new job id. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py` | +| Review pass 12 verification | completed | Re-ran focused ingestion/API/contract/retained E2E coverage, the public demo-script suite, and the repo functional smoke after narrowing the terminal cache. | `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 102 passed in 15.00s; `uv run pytest tests/scripts/test_demo_public_flow.py -q` -> 16 passed in 33.69s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 9.66s; `git diff --check` passed | +| Review pass 13 | completed | A fresh reviewer found two remaining race windows: callback-first cancelled background handoff could still skip the one-shot failed result, and the direct entrypoint only consumed that cache when a done task was still locally visible. | Findings referenced `indexing/ingestion_service.py` one-shot cache consumption order and missing callback-first regression coverage | +| Review pass 13 fixes | completed | Simplified direct `sync_source()` to reconcile then consume the one-shot cancelled-job cache on every entry, removed redundant cache writes from the reconcile path, and added a regression for the callback-first cancelled-background handoff. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py` | +| Review pass 13 verification | completed | Re-ran focused ingestion coverage, the broader ingestion/API/contract/retained E2E suite, the public demo-script suite, the repo functional smoke, and diff hygiene after the callback-first cancellation fix. | `uv run pytest tests/indexing/test_ingestion_service.py -q` -> 53 passed in 11.84s; `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 103 passed in 23.06s; `uv run pytest tests/scripts/test_demo_public_flow.py -q` -> 16 passed in 56.73s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 16.24s; `git diff --check` passed | +| Review pass 14 | completed | A fresh reviewer found one remaining authority gap: a cached cancelled local background job could still override a newer authoritative SQLite job started by another owner before the next direct `sync_source()` call. | Finding referenced `indexing/ingestion_service.py` cache freshness against the latest persisted job plus README launcher wording | +| Review pass 14 fixes | completed | Gated the one-shot cancelled-job cache on the latest persisted SQLite job still matching the cached cancelled job, added a regression for a newer foreign running job, and tightened README wording so the public MCP launcher no longer promises generic terminal-job returns. | `indexing/ingestion_service.py`; `tests/indexing/test_ingestion_service.py`; `README.md` | +| Review pass 14 verification | completed | Re-ran focused ingestion coverage, the broader ingestion/API/contract/retained E2E suite, the public demo-script suite, the repo functional smoke, and diff hygiene after the cache-authority fix. | `uv run pytest tests/indexing/test_ingestion_service.py -q` -> 54 passed in 3.26s; `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 104 passed in 10.66s; `uv run pytest tests/scripts/test_demo_public_flow.py -q` -> 16 passed in 21.74s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 7.01s; `git diff --check` passed | +| Review pass 15 | completed | A fresh reviewer found two remaining contract-maintenance gaps: the ADR/context note still described cancelled-local handoff too broadly, and the retained public contract coverage still did not pin the no-blocking-fallback branch when `start_sync_source()` is unavailable. | Findings referenced ADR 0007, `docs/contextwiki-core-understanding.md`, `tests/api/test_tools_contract.py`, and `tests/contracts/test_public_mcp_contracts.py` | +| Review pass 15 fixes | completed | Narrowed the ADR/context wording so cancelled-local handoff only applies while that cancelled job remains the latest authoritative SQLite job, and added explicit API plus retained public contract coverage for the no-background-launcher error path. | `.agents/docs/adr/0007-sync-source-background-launch-contract.md`; `docs/contextwiki-core-understanding.md`; `tests/api/test_tools_contract.py`; `tests/contracts/test_public_mcp_contracts.py` | +| Review pass 15 verification | completed | Re-ran focused API/contract coverage, the broader ingestion/API/contract/retained E2E suite, the public demo-script suite, the repo functional smoke, and diff hygiene after the contract-maintenance follow-up. | `uv run pytest tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py -q` -> 46 passed in 0.81s; `uv run pytest tests/indexing/test_ingestion_service.py tests/api/test_tools_contract.py tests/contracts/test_public_mcp_contracts.py tests/e2e/test_contextwiki_flow.py -q` -> 106 passed in 9.75s; `uv run pytest tests/scripts/test_demo_public_flow.py -q` -> 16 passed in 22.78s; `./scripts/verify_functional_e2e.sh` -> 25 passed in 7.17s; `git diff --check` passed | diff --git a/indexing/ingestion_service.py b/indexing/ingestion_service.py index a3f55cc..e2dc3e3 100644 --- a/indexing/ingestion_service.py +++ b/indexing/ingestion_service.py @@ -11,6 +11,8 @@ logger = logging.getLogger(__name__) +CANCELLED_SYNC_ERROR = "Sync request was cancelled before completion." + def _now() -> str: return datetime.now(timezone.utc).isoformat() @@ -67,6 +69,8 @@ def __init__( self.source_registry = source_registry self.chunker = chunker self.indexer = indexer + self._background_sync_tasks: dict[str, asyncio.Task] = {} + self._recent_terminal_background_jobs: dict[str, object] = {} self.register_source_config = register_source_config self.metadata_store.ensure_schema() if self.register_source_config: @@ -92,7 +96,10 @@ async def sync_all(self, source_ids: list[str] | None = None) -> dict: async def _sync_one(selected_source_id: str) -> dict: try: connector = self.source_registry.get_connector(selected_source_id) - job = await self.sync_source(selected_source_id) + job = await self._sync_source_internal( + selected_source_id, + join_existing_background=False, + ) except Exception as exc: message = _redact_sensitive_error(str(exc)) logger.error("Bulk sync failed for source %s: %s", selected_source_id, message) @@ -129,6 +136,85 @@ async def _sync_one(selected_source_id: str) -> dict: } async def sync_source(self, source_id: str): + self._reconcile_finished_background_task(source_id) + recent_terminal_job = self._recent_terminal_background_jobs.pop(source_id, None) + if recent_terminal_job is not None: + latest_job = self.metadata_store.get_latest_sync_job(source_id) + if ( + latest_job is not None + and latest_job.job_id == recent_terminal_job.job_id + and latest_job.status == SyncJobStatus.FAILED + and latest_job.error_message == CANCELLED_SYNC_ERROR + ): + return latest_job + return await self._sync_source_internal(source_id, join_existing_background=True) + + async def _sync_source_internal(self, source_id: str, *, join_existing_background: bool): + connector, job, should_run = self._begin_sync_source(source_id) + if not should_run: + existing_task = self._background_sync_tasks.get(source_id) + if ( + join_existing_background + and job.status == SyncJobStatus.RUNNING + and existing_task is not None + ): + try: + return await asyncio.shield(existing_task) + except asyncio.CancelledError: + if existing_task.cancelled() or existing_task.done(): + for _ in range(5): + self._reconcile_finished_background_task(source_id) + latest_job = self.metadata_store.get_latest_sync_job(source_id) + if latest_job is not None and latest_job.status != SyncJobStatus.RUNNING: + return latest_job + await asyncio.sleep(0) + raise + return job + return await self._run_sync_source_job(job.job_id, source_id, connector) + + async def start_sync_source(self, source_id: str): + self._reconcile_finished_background_task(source_id) + self._recent_terminal_background_jobs.pop(source_id, None) + connector, job, should_run = self._begin_sync_source(source_id) + if not should_run: + return job + try: + task = asyncio.create_task( + self._run_sync_source_job(job.job_id, source_id, connector), + name=f"sync-source:{source_id}:{job.job_id}", + ) + except Exception as exc: + error_message = _redact_sensitive_error(str(exc)) + logger.error( + "Unable to start background sync for source %s: %s", + source_id, + error_message, + ) + return self.metadata_store.complete_failed_sync( + job_id=job.job_id, + source_id=source_id, + error_message=error_message, + stale_cleanup_disabled_reason=( + _stale_cleanup_reason_for_connector(connector, error_message) + if not getattr(connector, "supports_stale_cleanup", False) + or not connector.source.enabled + else "" + ), + ) + self._background_sync_tasks[source_id] = task + setattr(task, "contextwiki_job_id", job.job_id) + setattr(task, "contextwiki_connector", connector) + task.add_done_callback( + lambda completed_task, sid=source_id, jid=job.job_id, sync_connector=connector: self._finalize_background_sync_task( + sid, + jid, + sync_connector, + completed_task, + ) + ) + return job + + def _begin_sync_source(self, source_id: str): connector = self.source_registry.get_connector(source_id) if self.register_source_config: self.refresh_registered_sources() @@ -137,12 +223,12 @@ async def sync_source(self, source_id: str): job, started = self.metadata_store.begin_sync_job(source_id) if not started: logger.info("Sync already running for source %s", source_id) - return job + return connector, job, False if not connector.source.enabled: message = _redact_sensitive_error( getattr(connector, "disabled_reason", "") or f"Source {source_id} is disabled" ) - return self.metadata_store.complete_failed_sync( + return connector, self.metadata_store.complete_failed_sync( job_id=job.job_id, source_id=source_id, error_message=message, @@ -150,9 +236,15 @@ async def sync_source(self, source_id: str): connector, message, ), - ) + ), False + return connector, job, True + async def _run_sync_source_job(self, job_id: str, source_id: str, connector): + job = None try: + job = self.metadata_store.get_sync_job(job_id) + if not job: + raise ValueError(f"Unknown sync job: {job_id}") inactive_job = self._refresh_running_job_or_current(job.job_id) if inactive_job: return inactive_job @@ -305,13 +397,30 @@ async def sync_source(self, source_id: str): await self._delete_vectors_best_effort(deleted_chunk_ids, source_id) return finished + except asyncio.CancelledError: + error_message = CANCELLED_SYNC_ERROR + logger.warning("Sync cancelled for source %s", source_id) + if "uncommitted_vector_ids" in locals(): + await self._delete_vectors_best_effort(uncommitted_vector_ids, source_id) + self.metadata_store.complete_failed_sync( + job_id=job.job_id, + source_id=source_id, + error_message=error_message, + stale_cleanup_disabled_reason=( + _stale_cleanup_reason_for_connector(connector, error_message) + if not getattr(connector, "supports_stale_cleanup", False) + or not connector.source.enabled + else "" + ), + ) + raise except Exception as exc: error_message = _redact_sensitive_error(str(exc)) logger.error("Sync failed for source %s: %s", source_id, error_message) if "uncommitted_vector_ids" in locals(): await self._delete_vectors_best_effort(uncommitted_vector_ids, source_id) return self.metadata_store.complete_failed_sync( - job_id=job.job_id, + job_id=job.job_id if job is not None else job_id, source_id=source_id, error_message=error_message, stale_cleanup_disabled_reason=( @@ -322,6 +431,50 @@ async def sync_source(self, source_id: str): ), ) + def _reconcile_finished_background_task(self, source_id: str) -> None: + existing_task = self._background_sync_tasks.get(source_id) + if existing_task is None or not existing_task.done(): + return + job_id = getattr(existing_task, "contextwiki_job_id", "") + connector = getattr(existing_task, "contextwiki_connector", None) + if not job_id or connector is None: + self._background_sync_tasks.pop(source_id, None) + return + self._finalize_background_sync_task(source_id, job_id, connector, existing_task) + + def _finalize_background_sync_task( + self, + source_id: str, + job_id: str, + connector, + task: asyncio.Task, + ) -> None: + current_task = self._background_sync_tasks.get(source_id) + if current_task is task: + self._background_sync_tasks.pop(source_id, None) + try: + task.result() + except asyncio.CancelledError: + logger.warning("Background sync task cancelled for source %s", source_id) + failed_job = self.metadata_store.complete_failed_sync( + job_id=job_id, + source_id=source_id, + error_message=CANCELLED_SYNC_ERROR, + stale_cleanup_disabled_reason=( + _stale_cleanup_reason_for_connector(connector, CANCELLED_SYNC_ERROR) + if not getattr(connector, "supports_stale_cleanup", False) + or not connector.source.enabled + else "" + ), + ) + self._recent_terminal_background_jobs[source_id] = failed_job + except Exception as exc: + logger.error( + "Background sync task failed for source %s: %s", + source_id, + _redact_sensitive_error(str(exc)), + ) + def _refresh_running_job_or_current(self, job_id: str): current_job = self.metadata_store.touch_sync_job(job_id) if not current_job: diff --git a/scripts/demo_public_flow.py b/scripts/demo_public_flow.py index a65f621..2f8ce1d 100644 --- a/scripts/demo_public_flow.py +++ b/scripts/demo_public_flow.py @@ -116,6 +116,22 @@ def build_demo_components(sample_vault: Path, temp_root: Path) -> DemoMCP: return mcp +async def _wait_for_demo_sync_completion( + mcp: DemoMCP, + source_id: str, + *, + attempts: int = 500, +) -> dict: + latest = None + for _ in range(attempts): + latest = await mcp.tools["get_sync_status"](source_id) + latest_job = latest.get("latest_job") or {} + if latest_job.get("status") in {"succeeded", "failed"}: + return latest + await asyncio.sleep(0.01) + raise AssertionError(f"Timed out waiting for demo sync completion: {latest}") + + async def run_demo(query: str, question: str) -> dict: had_embed_model_attr = hasattr(Settings, "_embed_model") previous_embed_model = getattr(Settings, "_embed_model", None) @@ -129,7 +145,7 @@ async def run_demo(query: str, question: str) -> dict: sample_vault = repo_root / "sample_vault" mcp = build_demo_components(sample_vault, Path(temp_dir)) sync_payload = await mcp.tools["sync_source"]("source_obsidian") - status_payload = await mcp.tools["get_sync_status"]("source_obsidian") + status_payload = await _wait_for_demo_sync_completion(mcp, "source_obsidian") search_payload = await mcp.tools["search_context"]( query, filters={"source_id": "source_obsidian"}, diff --git a/tests/api/test_tools_contract.py b/tests/api/test_tools_contract.py index 61a4366..e724792 100644 --- a/tests/api/test_tools_contract.py +++ b/tests/api/test_tools_contract.py @@ -55,12 +55,24 @@ async def index_documents(self, documents): class FakeFailingIngestion: - async def sync_source(self, source_id): + async def start_sync_source(self, source_id): raise ValueError(f"Unknown source: {source_id}") -class FakeLeakyJobIngestion: +class FakeBlockingOnlyIngestion: async def sync_source(self, source_id): + return Dumpable( + { + "job_id": "job-blocking-only", + "source_id": source_id, + "status": "succeeded", + "error_message": "", + } + ) + + +class FakeLeakyJobIngestion: + async def start_sync_source(self, source_id): return Dumpable( { "job_id": "job-leaky", @@ -226,7 +238,7 @@ async def sync_all(self): class FakePathFailingIngestion: - async def sync_source(self, source_id): + async def start_sync_source(self, source_id): raise ValueError( "Sync failed at /Users/eunhwa/private/vault.md " "with token supersecretvalue123456" @@ -664,6 +676,21 @@ def test_sync_source_redacts_returned_job_error_payload(): assert "ghp_secretcredential" not in payload +def test_sync_source_returns_error_when_background_launcher_is_unavailable(): + mcp = FakeMCP() + register_tools( + mcp, + ingestion_service=FakeBlockingOnlyIngestion(), + ) + + result = asyncio.run(mcp.tools["sync_source"]("source_github")) + + assert result == { + "status": "error", + "message": "ingestion service does not support background sync launch", + } + + def test_sync_source_redacts_public_error_paths_and_whitespace_secrets(): mcp = FakeMCP() register_tools( diff --git a/tests/contracts/test_public_mcp_contracts.py b/tests/contracts/test_public_mcp_contracts.py index eebef12..f198ecd 100644 --- a/tests/contracts/test_public_mcp_contracts.py +++ b/tests/contracts/test_public_mcp_contracts.py @@ -102,6 +102,9 @@ def __init__(self, source_registry: FakeSourceRegistry): chunk_id="chunk-1", ) + def set_job(self, source_id, job_payload): + self.jobs[source_id] = Dumpable(job_payload, **job_payload) + def register_source(self, source): self.sources[source.source_id] = source return source @@ -138,17 +141,29 @@ def list_chunks_for_document(self, document_id): class FakeIngestionService: - async def sync_source(self, source_id): - return Dumpable( - { - "job_id": f"job-{source_id}", - "source_id": source_id, - "status": "succeeded", - "started_at": "2026-06-15T00:00:00+00:00", - "finished_at": "2026-06-15T00:00:01+00:00", - "error_message": "", - } - ) + def __init__(self, metadata_store: FakeMetadataStore): + self.metadata_store = metadata_store + self.calls: dict[str, int] = {} + self.job_numbers: dict[str, int] = {} + + async def start_sync_source(self, source_id): + self.calls[source_id] = self.calls.get(source_id, 0) + 1 + existing_job = self.metadata_store.get_latest_sync_job(source_id) + if existing_job and getattr(existing_job, "status", "") == "running": + return existing_job + + next_job_number = self.job_numbers.get(source_id, 0) + 1 + self.job_numbers[source_id] = next_job_number + job_payload = { + "job_id": f"job-{source_id}-{next_job_number}", + "source_id": source_id, + "status": "running", + "started_at": "2026-06-15T00:00:00+00:00", + "finished_at": "", + "error_message": "", + } + self.metadata_store.set_job(source_id, job_payload) + return Dumpable(job_payload, **job_payload) async def sync_all(self): return { @@ -195,6 +210,26 @@ async def sync_all(self): } +class FakeBlockingOnlyIngestionService: + async def sync_source(self, source_id): + return Dumpable( + { + "job_id": f"job-{source_id}-blocking-only", + "source_id": source_id, + "status": "succeeded", + "started_at": "2026-06-15T00:00:00+00:00", + "finished_at": "2026-06-15T00:00:01+00:00", + "error_message": "", + }, + job_id=f"job-{source_id}-blocking-only", + source_id=source_id, + status="succeeded", + started_at="2026-06-15T00:00:00+00:00", + finished_at="2026-06-15T00:00:01+00:00", + error_message="", + ) + + class FakeContextSearchService: async def search_context(self, query, filters=None, top_k=10, include_debug=False): debug_payload = {} @@ -271,19 +306,28 @@ async def answer_with_citations(self, question, filters=None, top_k=5, include_d return payload -def build_contract_mcp() -> FastMCP: +def build_contract_harness(): source_registry = FakeSourceRegistry() metadata_store = FakeMetadataStore(source_registry) mcp = FastMCP("public-contract-test") + ingestion_service = FakeIngestionService(metadata_store) register_tools( mcp, - ingestion_service=FakeIngestionService(), + ingestion_service=ingestion_service, context_search_service=FakeContextSearchService(), answer_service=FakeAnswerService(), metadata_store=metadata_store, source_registry=source_registry, ) - return mcp + return { + "mcp": mcp, + "metadata_store": metadata_store, + "ingestion_service": ingestion_service, + } + + +def build_contract_mcp() -> FastMCP: + return build_contract_harness()["mcp"] def call_tool_json(mcp: FastMCP, name: str, arguments: dict | None = None) -> dict: @@ -308,10 +352,62 @@ def test_sync_source_contract_uses_real_fastmcp_call_tool(): {"source_id": "source_github"}, ) - assert payload["status"] == "succeeded" + assert payload["status"] == "running" assert payload["source_id"] == "source_github" +def test_sync_source_contract_reuses_running_job_and_polls_to_terminal_status(): + harness = build_contract_harness() + mcp = harness["mcp"] + metadata_store = harness["metadata_store"] + ingestion_service = harness["ingestion_service"] + + first_payload = call_tool_json(mcp, "sync_source", {"source_id": "source_github"}) + second_payload = call_tool_json(mcp, "sync_source", {"source_id": "source_github"}) + + assert first_payload["status"] == "running" + assert second_payload["status"] == "running" + assert second_payload["job_id"] == first_payload["job_id"] + assert ingestion_service.calls["source_github"] == 2 + + metadata_store.set_job( + "source_github", + { + "job_id": first_payload["job_id"], + "source_id": "source_github", + "status": "succeeded", + "started_at": first_payload["started_at"], + "finished_at": "2026-06-15T00:00:01+00:00", + "error_message": "", + }, + ) + status_payload = call_tool_json(mcp, "get_sync_status", {"source_id": "source_github"}) + + assert status_payload["latest_job"]["job_id"] == first_payload["job_id"] + assert status_payload["latest_job"]["status"] == "succeeded" + + +def test_sync_source_contract_returns_error_without_background_launcher(): + source_registry = FakeSourceRegistry() + metadata_store = FakeMetadataStore(source_registry) + mcp = FastMCP("public-contract-test-no-launcher") + register_tools( + mcp, + ingestion_service=FakeBlockingOnlyIngestionService(), + context_search_service=FakeContextSearchService(), + answer_service=FakeAnswerService(), + metadata_store=metadata_store, + source_registry=source_registry, + ) + + payload = call_tool_json(mcp, "sync_source", {"source_id": "source_github"}) + + assert payload == { + "status": "error", + "message": "ingestion service does not support background sync launch", + } + + def test_sync_all_contract_uses_real_fastmcp_call_tool(): payload = call_tool_json(build_contract_mcp(), "sync_all") diff --git a/tests/e2e/test_contextwiki_flow.py b/tests/e2e/test_contextwiki_flow.py index f707fab..f762ceb 100644 --- a/tests/e2e/test_contextwiki_flow.py +++ b/tests/e2e/test_contextwiki_flow.py @@ -117,6 +117,29 @@ def _call_tool_json(mcp: FastMCP, name: str, arguments: dict | None = None) -> d return json.loads(blocks[0].text) +async def _call_tool_json_async(mcp: FastMCP, name: str, arguments: dict | None = None) -> dict: + blocks = await mcp.call_tool(name, arguments or {}) + return json.loads(blocks[0].text) + + +async def _wait_for_sync_completion(mcp, source_id: str, attempts: int = 500) -> dict: + latest = None + for _ in range(attempts): + if hasattr(mcp, "tools"): + latest = await mcp.tools["get_sync_status"](source_id) + else: + latest = await _call_tool_json_async( + mcp, + "get_sync_status", + {"source_id": source_id}, + ) + latest_job = latest.get("latest_job") or {} + if latest_job.get("status") in {"succeeded", "failed"}: + return latest + await asyncio.sleep(0.01) + raise AssertionError(f"Timed out waiting for {source_id} sync completion: {latest}") + + pytestmark = pytest.mark.e2e @@ -143,36 +166,54 @@ def test_contextwiki_fake_e2e_sync_search_fetch_and_answer(tmp_path): source_registry=registry, ) - sync_job = asyncio.run(mcp.tools["sync_source"]("source_fake_docs")) - status = asyncio.run(mcp.tools["get_sync_status"]("source_fake_docs")) - search_result = asyncio.run( - mcp.tools["search_context"]( + async def run_flow(): + sync_job = await mcp.tools["sync_source"]("source_fake_docs") + status = await _wait_for_sync_completion(mcp, "source_fake_docs") + search_result = await mcp.tools["search_context"]( "citations", filters={"source_ids": ["source_fake_docs"]}, top_k=5, ) - ) - collection_search = asyncio.run( - mcp.tools["search_context"]( + collection_search = await mcp.tools["search_context"]( "ContextWiki 관련 문서 모아줘", filters={"source_ids": ["source_fake_docs"]}, top_k=5, include_debug=True, ) - ) - chunk_id = search_result["results"][0]["chunk_id"] - fetched = asyncio.run(mcp.tools["fetch_context"](chunk_id=chunk_id)) - answer = asyncio.run(mcp.tools["answer_with_citations"]("How does ContextWiki answer?")) - collection_answer = asyncio.run( - mcp.tools["answer_with_citations"]( + chunk_id = search_result["results"][0]["chunk_id"] + fetched = await mcp.tools["fetch_context"](chunk_id=chunk_id) + answer = await mcp.tools["answer_with_citations"]("How does ContextWiki answer?") + collection_answer = await mcp.tools["answer_with_citations"]( "ContextWiki 관련 문서 모아줘", filters={"source_ids": ["source_fake_docs"]}, top_k=5, ) - ) - unsupported = asyncio.run(mcp.tools["answer_with_citations"]("What is the deployment region?")) + unsupported = await mcp.tools["answer_with_citations"]("What is the deployment region?") + return ( + sync_job, + status, + search_result, + collection_search, + chunk_id, + fetched, + answer, + collection_answer, + unsupported, + ) - assert sync_job["status"] == "succeeded" + ( + sync_job, + status, + search_result, + collection_search, + chunk_id, + fetched, + answer, + collection_answer, + unsupported, + ) = asyncio.run(run_flow()) + + assert sync_job["status"] == "running" assert status["source"]["sync_status"] == "succeeded" assert search_result["results"][0]["title"] == "ContextWiki MVP" assert collection_search["debug"]["intent"]["name"] == "list" @@ -258,29 +299,32 @@ def test_contextwiki_temp_chroma_e2e_sync_search_fetch_and_answer(tmp_path): ] ) ) - other_job = asyncio.run(mcp.tools["sync_source"]("source_other")) - target_job = asyncio.run(mcp.tools["sync_source"]("source_fake_docs")) - status = asyncio.run(mcp.tools["get_sync_status"]("source_fake_docs")) - search_result = asyncio.run( - mcp.tools["search_context"]( + async def run_flow(): + other_job = await mcp.tools["sync_source"]("source_other") + target_job = await mcp.tools["sync_source"]("source_fake_docs") + await _wait_for_sync_completion(mcp, "source_other") + status = await _wait_for_sync_completion(mcp, "source_fake_docs") + search_result = await mcp.tools["search_context"]( "ContextWiki citations", filters={"source_id": "source_fake_docs"}, top_k=1, ) - ) - chunk_id = search_result["results"][0]["chunk_id"] - fetched = asyncio.run(mcp.tools["fetch_context"](chunk_id=chunk_id)) - answer = asyncio.run( - mcp.tools["answer_with_citations"]( + chunk_id = search_result["results"][0]["chunk_id"] + fetched = await mcp.tools["fetch_context"](chunk_id=chunk_id) + answer = await mcp.tools["answer_with_citations"]( "How does ContextWiki answer?", filters={"source_id": "source_fake_docs"}, top_k=1, ) + return other_job, target_job, status, search_result, chunk_id, fetched, answer + + other_job, target_job, status, search_result, chunk_id, fetched, answer = asyncio.run( + run_flow() ) metadatas = chroma_collection.get(include=["metadatas"])["metadatas"] - assert other_job["status"] == "succeeded" - assert target_job["status"] == "succeeded" + assert other_job["status"] == "running" + assert target_job["status"] == "running" assert status["source"]["sync_status"] == "succeeded" assert chroma_collection.count() >= 3 assert any(metadata.get("contextwiki_managed") == "false" for metadata in metadatas) @@ -359,7 +403,12 @@ def retrieve(self, query): source_registry=registry, ) - sync_job = _call_tool_json(mcp, "sync_source", {"source_id": "source_alias_docs"}) + async def run_flow(): + sync_job = await _call_tool_json_async(mcp, "sync_source", {"source_id": "source_alias_docs"}) + await _wait_for_sync_completion(mcp, "source_alias_docs") + return sync_job + + sync_job = asyncio.run(run_flow()) chunk_id = store.list_chunks_for_document("doc_aws_alias")[0].chunk_id search_result = _call_tool_json( mcp, @@ -371,7 +420,7 @@ def retrieve(self, query): }, ) - assert sync_job["status"] == "succeeded" + assert sync_job["status"] == "running" assert len(search_result["results"]) == 1 assert search_result["results"][0]["title"] == "Cloud deployment checklist" assert search_result["results"][0]["chunk_id"] == chunk_id @@ -459,7 +508,16 @@ def retrieve(self, query): source_registry=registry, ) - sync_job = _call_tool_json(rewrite_mcp, "sync_source", {"source_id": "source_rewrite_docs"}) + async def run_flow(): + sync_job = await _call_tool_json_async( + rewrite_mcp, + "sync_source", + {"source_id": "source_rewrite_docs"}, + ) + await _wait_for_sync_completion(rewrite_mcp, "source_rewrite_docs") + return sync_job + + sync_job = asyncio.run(run_flow()) chunk_id = store.list_chunks_for_document("doc_ec2_setup")[0].chunk_id baseline_result = _call_tool_json( baseline_mcp, @@ -482,7 +540,7 @@ def retrieve(self, query): }, ) - assert sync_job["status"] == "succeeded" + assert sync_job["status"] == "running" assert baseline_result["results"] == [] assert "aws virtual machine startup" in baseline_queries assert "aws ec2 setup" not in baseline_queries @@ -604,7 +662,16 @@ def retrieve(self, query): source_registry=registry, ) - sync_job = _call_tool_json(mcp, "sync_source", {"source_id": "source_github_docs_intent"}) + async def run_flow(): + sync_job = await _call_tool_json_async( + mcp, + "sync_source", + {"source_id": "source_github_docs_intent"}, + ) + await _wait_for_sync_completion(mcp, "source_github_docs_intent") + return sync_job + + sync_job = asyncio.run(run_flow()) code_document_count = sum( 1 for document in indexer.documents @@ -626,7 +693,7 @@ def retrieve(self, query): }, ) - assert sync_job["status"] == "succeeded" + assert sync_job["status"] == "running" assert code_document_count == 64 assert retrieved_queries assert returned_candidate_ids[0] == docs_chunk_id diff --git a/tests/e2e/test_obsidian_connector_flow.py b/tests/e2e/test_obsidian_connector_flow.py index b6fd4c4..e89c426 100644 --- a/tests/e2e/test_obsidian_connector_flow.py +++ b/tests/e2e/test_obsidian_connector_flow.py @@ -49,6 +49,22 @@ def _call_tool_json(mcp: FastMCP, name: str, arguments: dict | None = None) -> d return json.loads(blocks[0].text) +async def _call_tool_json_async(mcp: FastMCP, name: str, arguments: dict | None = None) -> dict: + blocks = await mcp.call_tool(name, arguments or {}) + return json.loads(blocks[0].text) + + +async def _wait_for_sync_completion(mcp: FastMCP, source_id: str, attempts: int = 500) -> dict: + latest = None + for _ in range(attempts): + latest = await _call_tool_json_async(mcp, "get_sync_status", {"source_id": source_id}) + latest_job = latest.get("latest_job") or {} + if latest_job.get("status") in {"succeeded", "failed"}: + return latest + await asyncio.sleep(0.01) + raise AssertionError(f"Timed out waiting for {source_id} sync completion: {latest}") + + def _make_vault(tmp_path, files: dict[str, str]): vault = tmp_path / "vault" vault.mkdir() @@ -118,9 +134,13 @@ def test_obsidian_sync_through_mcp_tools_indexes_temp_vault_notes_with_citations connector, store, indexer, mcp = _obsidian_service(tmp_path, vault) - listed = _call_tool_json(mcp, "list_sources") - sync_job = _call_tool_json(mcp, "sync_source", {"source_id": "source_obsidian"}) - status = _call_tool_json(mcp, "get_sync_status", {"source_id": "source_obsidian"}) + async def run_flow(): + listed = await _call_tool_json_async(mcp, "list_sources") + sync_job = await _call_tool_json_async(mcp, "sync_source", {"source_id": "source_obsidian"}) + status = await _wait_for_sync_completion(mcp, "source_obsidian") + return listed, sync_job, status + + listed, sync_job, status = asyncio.run(run_flow()) project = store.get_document("notes/project.md") project_chunks = store.list_chunks_for_document("notes/project.md") search_result = _call_tool_json( @@ -155,11 +175,11 @@ def test_obsidian_sync_through_mcp_tools_indexes_temp_vault_notes_with_citations assert [source["source_id"] for source in listed["sources"]] == ["source_obsidian"] assert listed["sources"][0]["enabled"] is True assert connector.supports_stale_cleanup is True - assert sync_job["status"] == "succeeded" + assert sync_job["status"] == "running" assert sync_job["source_id"] == "source_obsidian" - assert sync_job["processed_documents"] == 2 assert status["source"]["sync_status"] == "succeeded" assert status["latest_job"]["status"] == "succeeded" + assert status["latest_job"]["processed_documents"] == 2 assert store.get_source("source_obsidian").sync_status == SyncStatus.SUCCEEDED assert project is not None assert project.source_id == "source_obsidian" diff --git a/tests/e2e/test_phase_b_connectors_flow.py b/tests/e2e/test_phase_b_connectors_flow.py index 05ced74..0081528 100644 --- a/tests/e2e/test_phase_b_connectors_flow.py +++ b/tests/e2e/test_phase_b_connectors_flow.py @@ -60,6 +60,22 @@ def _call_tool_json(mcp: FastMCP, name: str, arguments: dict | None = None) -> d return json.loads(blocks[0].text) +async def _call_tool_json_async(mcp: FastMCP, name: str, arguments: dict | None = None) -> dict: + blocks = await mcp.call_tool(name, arguments or {}) + return json.loads(blocks[0].text) + + +async def _wait_for_sync_completion(mcp: FastMCP, source_id: str, attempts: int = 500) -> dict: + latest = None + for _ in range(attempts): + latest = await _call_tool_json_async(mcp, "get_sync_status", {"source_id": source_id}) + latest_job = latest.get("latest_job") or {} + if latest_job.get("status") in {"succeeded", "failed"}: + return latest + await asyncio.sleep(0.01) + raise AssertionError(f"Timed out waiting for {source_id} sync completion: {latest}") + + class FakeRetainedSourceConnector(SourceConnector): def __init__( self, @@ -366,9 +382,13 @@ def test_retained_notion_and_tistory_sync_through_mcp_tools( source_registry=registry, ) - listed = _call_tool_json(mcp, "list_sources") - sync_job = _call_tool_json(mcp, "sync_source", {"source_id": source_id}) - status = _call_tool_json(mcp, "get_sync_status", {"source_id": source_id}) + async def run_flow(): + listed = await _call_tool_json_async(mcp, "list_sources") + sync_job = await _call_tool_json_async(mcp, "sync_source", {"source_id": source_id}) + status = await _wait_for_sync_completion(mcp, source_id) + return listed, sync_job, status + + listed, sync_job, status = asyncio.run(run_flow()) document_id = connectors[source_id]._document.document_id chunks = store.list_chunks_for_document(document_id) search_result = _call_tool_json( @@ -404,12 +424,12 @@ def test_retained_notion_and_tistory_sync_through_mcp_tools( "source_notion", "source_tistory", ] - assert sync_job["status"] == "succeeded" + assert sync_job["status"] == "running" assert sync_job["source_id"] == source_id - assert sync_job["processed_documents"] == 1 - assert sync_job["indexed_chunks"] >= 1 assert status["source"]["sync_status"] == "succeeded" assert status["latest_job"]["status"] == "succeeded" + assert status["latest_job"]["processed_documents"] == 1 + assert status["latest_job"]["indexed_chunks"] >= 1 assert connectors[source_id].fetch_count == 1 assert all( connector.fetch_count == (1 if connector.source.source_id == source_id else 0) @@ -460,9 +480,13 @@ def test_retained_github_sync_through_mcp_tools(tmp_path): source_registry=registry, ) - listed = _call_tool_json(mcp, "list_sources") - sync_job = _call_tool_json(mcp, "sync_source", {"source_id": "source_github"}) - status = _call_tool_json(mcp, "get_sync_status", {"source_id": "source_github"}) + async def run_flow(): + listed = await _call_tool_json_async(mcp, "list_sources") + sync_job = await _call_tool_json_async(mcp, "sync_source", {"source_id": "source_github"}) + status = await _wait_for_sync_completion(mcp, "source_github") + return listed, sync_job, status + + listed, sync_job, status = asyncio.run(run_flow()) document_id = "github:eunhwa99/mcpcontentsearch:api/tools.py" chunks = store.list_chunks_for_document(document_id) search_result = _call_tool_json( @@ -495,12 +519,12 @@ def test_retained_github_sync_through_mcp_tools(tmp_path): ) assert [source["source_id"] for source in listed["sources"]] == ["source_github"] - assert sync_job["status"] == "succeeded" + assert sync_job["status"] == "running" assert sync_job["source_id"] == "source_github" - assert sync_job["processed_documents"] == 1 - assert sync_job["indexed_chunks"] >= 1 assert status["source"]["sync_status"] == "succeeded" assert status["latest_job"]["status"] == "succeeded" + assert status["latest_job"]["processed_documents"] == 1 + assert status["latest_job"]["indexed_chunks"] >= 1 assert store.get_source("source_github").sync_status == SyncStatus.SUCCEEDED assert store.get_document(document_id).source_id == "source_github" assert chunks diff --git a/tests/indexing/test_ingestion_service.py b/tests/indexing/test_ingestion_service.py index 889e775..e7bc719 100644 --- a/tests/indexing/test_ingestion_service.py +++ b/tests/indexing/test_ingestion_service.py @@ -378,6 +378,571 @@ async def run_overlapping_syncs(): assert store.get_source("source_fake").sync_status == SyncStatus.SUCCEEDED +def test_start_sync_source_returns_running_job_and_completes_in_background(tmp_path): + document = DocumentModel( + id="doc-started", + source_id="source_fake", + title="Background", + content="Background launch should return before completion.", + url="https://example.com/doc-started", + platform="GitHub", + path="doc-started.md", + ) + + async def run_background_launch(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + await started.wait() + running_job = store.get_latest_sync_job("source_fake") + release.set() + + completed_job = None + for _ in range(20): + completed_job = store.get_latest_sync_job("source_fake") + if completed_job and completed_job.status != SyncJobStatus.RUNNING: + break + await asyncio.sleep(0) + else: + raise AssertionError("background sync did not complete within polling window") + + return connector, launched_job, running_job, completed_job, store + + connector, launched_job, running_job, completed_job, store = asyncio.run( + run_background_launch() + ) + + assert connector.calls == 1 + assert launched_job.status == SyncJobStatus.RUNNING + assert running_job is not None + assert running_job.status == SyncJobStatus.RUNNING + assert completed_job is not None + assert completed_job.job_id == launched_job.job_id + assert completed_job.status == SyncJobStatus.SUCCEEDED + assert store.get_source("source_fake").sync_status == SyncStatus.SUCCEEDED + + +def test_start_sync_source_reuses_existing_running_job_without_second_fetch(tmp_path): + document = DocumentModel( + id="doc-reuse", + source_id="source_fake", + title="Reuse", + content="A second launcher call should reuse the running job.", + url="https://example.com/doc-reuse", + platform="GitHub", + path="doc-reuse.md", + ) + + async def run_reused_background_launch(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + first_job = await service.start_sync_source("source_fake") + await started.wait() + second_job = await service.start_sync_source("source_fake") + release.set() + + completed_job = None + for _ in range(20): + completed_job = store.get_latest_sync_job("source_fake") + if completed_job and completed_job.status != SyncJobStatus.RUNNING: + break + await asyncio.sleep(0) + else: + raise AssertionError("background sync did not complete within polling window") + + return connector, first_job, second_job, completed_job + + connector, first_job, second_job, completed_job = asyncio.run( + run_reused_background_launch() + ) + + assert connector.calls == 1 + assert first_job.status == SyncJobStatus.RUNNING + assert second_job.status == SyncJobStatus.RUNNING + assert second_job.job_id == first_job.job_id + assert completed_job is not None + assert completed_job.job_id == first_job.job_id + assert completed_job.status == SyncJobStatus.SUCCEEDED + + +def test_sync_source_waits_for_local_background_job_completion(tmp_path): + document = DocumentModel( + id="doc-blocking-join", + source_id="source_fake", + title="Blocking join", + content="Direct sync_source should wait for a locally launched background sync.", + url="https://example.com/doc-blocking-join", + platform="GitHub", + path="doc-blocking-join.md", + ) + + async def run_blocking_join(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + await started.wait() + direct_task = asyncio.create_task(service.sync_source("source_fake")) + await asyncio.sleep(0) + assert not direct_task.done() + release.set() + completed_job = await direct_task + return connector, launched_job, completed_job, store + + connector, launched_job, completed_job, store = asyncio.run(run_blocking_join()) + + assert connector.calls == 1 + assert launched_job.status == SyncJobStatus.RUNNING + assert completed_job.job_id == launched_job.job_id + assert completed_job.status == SyncJobStatus.SUCCEEDED + assert store.get_latest_sync_job("source_fake").status == SyncJobStatus.SUCCEEDED + + +def test_sync_source_starts_fresh_run_after_successful_background_completion(tmp_path): + document = DocumentModel( + id="doc-background-success-rerun", + source_id="source_fake", + title="Background rerun", + content="A completed background sync should not be cached as the next direct sync result.", + url="https://example.com/doc-background-success-rerun", + platform="GitHub", + path="doc-background-success-rerun.md", + ) + + async def run_background_then_direct_sync(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + await started.wait() + release.set() + for _ in range(20): + latest_job = store.get_latest_sync_job("source_fake") + if latest_job and latest_job.status == SyncJobStatus.SUCCEEDED: + break + await asyncio.sleep(0) + rerun_job = await service.sync_source("source_fake") + return connector, launched_job, rerun_job + + connector, launched_job, rerun_job = asyncio.run(run_background_then_direct_sync()) + + assert connector.calls == 2 + assert launched_job.status == SyncJobStatus.RUNNING + assert rerun_job.status == SyncJobStatus.SUCCEEDED + assert rerun_job.job_id != launched_job.job_id + + +def test_sync_source_returns_failed_job_when_joined_background_task_is_cancelled(tmp_path): + document = DocumentModel( + id="doc-blocking-cancel", + source_id="source_fake", + title="Blocking cancel", + content="Direct sync_source should return the reconciled failed job when the joined background task is cancelled.", + url="https://example.com/doc-blocking-cancel", + platform="GitHub", + path="doc-blocking-cancel.md", + ) + + async def run_blocking_join_cancel(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + await started.wait() + background_task = service._background_sync_tasks["source_fake"] + direct_task = asyncio.create_task(service.sync_source("source_fake")) + await asyncio.sleep(0) + background_task.cancel() + completed_job = await direct_task + release.set() + return launched_job, completed_job, store + + launched_job, completed_job, store = asyncio.run(run_blocking_join_cancel()) + + assert launched_job.status == SyncJobStatus.RUNNING + assert completed_job.job_id == launched_job.job_id + assert completed_job.status == SyncJobStatus.FAILED + assert completed_job.error_message == "Sync request was cancelled before completion." + assert store.get_latest_sync_job("source_fake").status == SyncJobStatus.FAILED + + +def test_sync_source_returns_failed_job_when_joined_background_task_is_cancelled_before_start( + tmp_path, +): + document = DocumentModel( + id="doc-blocking-prestart-cancel", + source_id="source_fake", + title="Blocking prestart cancel", + content="Direct sync_source should return the reconciled failed job even when the joined background task is cancelled before it starts.", + url="https://example.com/doc-blocking-prestart-cancel", + platform="GitHub", + path="doc-blocking-prestart-cancel.md", + ) + + async def run_blocking_join_prestart_cancel(): + connector = FakeConnector([document]) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + background_task = service._background_sync_tasks["source_fake"] + direct_task = asyncio.create_task(service.sync_source("source_fake")) + background_task.cancel() + completed_job = await direct_task + return launched_job, completed_job, store + + launched_job, completed_job, store = asyncio.run(run_blocking_join_prestart_cancel()) + + assert launched_job.status == SyncJobStatus.RUNNING + assert completed_job.job_id == launched_job.job_id + assert completed_job.status == SyncJobStatus.FAILED + assert completed_job.error_message == "Sync request was cancelled before completion." + assert store.get_latest_sync_job("source_fake").status == SyncJobStatus.FAILED + + +def test_cancelled_background_sync_task_marks_job_failed(tmp_path): + document = DocumentModel( + id="doc-cancel-window", + source_id="source_fake", + title="Cancelled background", + content="Cancelling the background task should not leave a stuck running job.", + url="https://example.com/doc-cancel-window", + platform="GitHub", + path="doc-cancel-window.md", + ) + + async def run_cancelled_background_task(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + background_task = service._background_sync_tasks["source_fake"] + background_task.cancel() + with pytest.raises(asyncio.CancelledError): + await background_task + await asyncio.sleep(0) + failed_job = store.get_latest_sync_job("source_fake") + release.set() + return launched_job, failed_job, store + + launched_job, failed_job, store = asyncio.run(run_cancelled_background_task()) + + assert launched_job.status == SyncJobStatus.RUNNING + assert failed_job is not None + assert failed_job.job_id == launched_job.job_id + assert failed_job.status == SyncJobStatus.FAILED + assert failed_job.error_message == "Sync request was cancelled before completion." + assert store.get_source("source_fake").sync_status == SyncStatus.FAILED + + +def test_start_sync_source_immediately_retries_after_cancelled_background_task(tmp_path): + document = DocumentModel( + id="doc-cancel-retry", + source_id="source_fake", + title="Cancel retry", + content="A cancelled local background sync should not block an immediate retry.", + url="https://example.com/doc-cancel-retry", + platform="GitHub", + path="doc-cancel-retry.md", + ) + + async def run_cancel_then_immediate_retry(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + first_job = await service.start_sync_source("source_fake") + await started.wait() + background_task = service._background_sync_tasks["source_fake"] + background_task.cancel() + with pytest.raises(asyncio.CancelledError): + await background_task + retried_job = await service.start_sync_source("source_fake") + release.set() + + for _ in range(20): + latest_job = store.get_latest_sync_job("source_fake") + if latest_job and latest_job.job_id == retried_job.job_id and latest_job.status != SyncJobStatus.RUNNING: + return connector, first_job, retried_job, latest_job + await asyncio.sleep(0) + raise AssertionError("immediate retry did not reach a terminal status") + + connector, first_job, retried_job, latest_job = asyncio.run(run_cancel_then_immediate_retry()) + + assert connector.calls == 2 + assert first_job.status == SyncJobStatus.RUNNING + assert retried_job.status == SyncJobStatus.RUNNING + assert retried_job.job_id != first_job.job_id + assert latest_job.job_id == retried_job.job_id + assert latest_job.status == SyncJobStatus.SUCCEEDED + + +def test_sync_source_returns_cancelled_background_failure_once_after_callback_finalizes(tmp_path): + document = DocumentModel( + id="doc-cancelled-callback-handoff", + source_id="source_fake", + title="Cancelled callback handoff", + content="A later direct sync should surface the reconciled cancelled background job once even after the callback finalizes it.", + url="https://example.com/doc-cancelled-callback-handoff", + platform="GitHub", + path="doc-cancelled-callback-handoff.md", + ) + + async def run_cancel_then_direct_sync(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + await started.wait() + background_task = service._background_sync_tasks["source_fake"] + background_task.cancel() + with pytest.raises(asyncio.CancelledError): + await background_task + await asyncio.sleep(0) + + failed_job = await service.sync_source("source_fake") + release.set() + rerun_job = await service.sync_source("source_fake") + return connector, launched_job, failed_job, rerun_job, store + + connector, launched_job, failed_job, rerun_job, store = asyncio.run(run_cancel_then_direct_sync()) + + assert connector.calls == 2 + assert failed_job.job_id == launched_job.job_id + assert failed_job.status == SyncJobStatus.FAILED + assert failed_job.error_message == "Sync request was cancelled before completion." + assert rerun_job.status == SyncJobStatus.SUCCEEDED + assert rerun_job.job_id != launched_job.job_id + assert store.get_latest_sync_job("source_fake").job_id == rerun_job.job_id + + +def test_sync_source_ignores_cancelled_background_cache_when_newer_foreign_job_is_running(tmp_path): + document = DocumentModel( + id="doc-cancelled-foreign-running", + source_id="source_fake", + title="Cancelled foreign running", + content="A stale cancelled background handoff must not override a newer authoritative running job.", + url="https://example.com/doc-cancelled-foreign-running", + platform="GitHub", + path="doc-cancelled-foreign-running.md", + ) + + async def run_cancel_then_foreign_restart(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + launched_job = await service.start_sync_source("source_fake") + await started.wait() + background_task = service._background_sync_tasks["source_fake"] + background_task.cancel() + with pytest.raises(asyncio.CancelledError): + await background_task + await asyncio.sleep(0) + + foreign_job, started_foreign = store.begin_sync_job("source_fake") + assert started_foreign is True + direct_job = await service.sync_source("source_fake") + release.set() + return launched_job, foreign_job, direct_job + + launched_job, foreign_job, direct_job = asyncio.run(run_cancel_then_foreign_restart()) + + assert direct_job.job_id != launched_job.job_id + assert direct_job.job_id == foreign_job.job_id + assert direct_job.status == SyncJobStatus.RUNNING + + +def test_background_sync_missing_initial_job_marks_job_failed(tmp_path, monkeypatch): + document = DocumentModel( + id="doc-missing-job", + source_id="source_fake", + title="Missing job", + content="Initial metadata lookup failures should not leave the job running.", + url="https://example.com/doc-missing-job", + platform="GitHub", + path="doc-missing-job.md", + ) + + async def run_missing_job_flow(): + connector = FakeConnector([document]) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + original_get_sync_job = store.get_sync_job + first_lookup = True + + def fail_first_get_sync_job(job_id): + nonlocal first_lookup + if first_lookup: + first_lookup = False + return None + return original_get_sync_job(job_id) + + monkeypatch.setattr(store, "get_sync_job", fail_first_get_sync_job) + + launched_job = await service.start_sync_source("source_fake") + + for _ in range(20): + failed_job = store.get_latest_sync_job("source_fake") + if failed_job and failed_job.status != SyncJobStatus.RUNNING: + return launched_job, failed_job, store + await asyncio.sleep(0) + raise AssertionError("missing-job failure did not reach a terminal status") + + launched_job, failed_job, store = asyncio.run(run_missing_job_flow()) + + assert launched_job.status == SyncJobStatus.RUNNING + assert failed_job is not None + assert failed_job.job_id == launched_job.job_id + assert failed_job.status == SyncJobStatus.FAILED + assert failed_job.error_message == "Unknown sync job: " + launched_job.job_id + assert store.get_source("source_fake").sync_status == SyncStatus.FAILED + + +def test_cancelled_source_sync_marks_job_failed_and_allows_retry(tmp_path): + document = DocumentModel( + id="doc-cancelled", + source_id="source_fake", + title="Cancelled", + content="A cancelled sync should not stay running forever.", + url="https://example.com/doc-cancelled", + platform="GitHub", + path="doc-cancelled.md", + ) + + async def run_cancelled_sync(): + started = asyncio.Event() + release = asyncio.Event() + connector = BlockingConnector([document], started, release) + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + indexer = RecordingIndexer() + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([connector]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=indexer, + ) + + first_task = asyncio.create_task(service.sync_source("source_fake")) + await started.wait() + first_task.cancel() + with pytest.raises(asyncio.CancelledError): + await first_task + failed_job = store.get_latest_sync_job("source_fake") + source_after_cancel = store.get_source("source_fake") + release.set() + retried_job = await service.sync_source("source_fake") + return connector, failed_job, source_after_cancel, retried_job, store + + connector, failed_job, source_after_cancel, retried_job, store = asyncio.run( + run_cancelled_sync() + ) + + assert connector.calls == 2 + assert failed_job is not None + assert failed_job.status == SyncJobStatus.FAILED + assert failed_job.error_message == "Sync request was cancelled before completion." + assert source_after_cancel is not None + assert source_after_cancel.sync_status == SyncStatus.FAILED + assert source_after_cancel.last_error == "Sync request was cancelled before completion." + assert retried_job.status == SyncJobStatus.SUCCEEDED + assert store.get_latest_sync_job("source_fake").status == SyncJobStatus.SUCCEEDED + + def test_source_registration_preserves_existing_sync_status(tmp_path): store = MetadataStore(tmp_path / "contextwiki.sqlite3") connector = FakeConnector() @@ -757,6 +1322,54 @@ async def run_sync_all_while_one_source_is_running(): assert result["summary"]["succeeded"] == 1 +def test_sync_all_reports_blocked_for_local_background_sync(tmp_path): + document = DocumentModel( + id="doc-a", + source_id="source_a", + title="Source A", + content="source a content", + url="https://example.com/a", + platform="GitHub", + path="a.md", + ) + + async def run_sync_all_while_background_sync_is_running(): + started = asyncio.Event() + release = asyncio.Event() + store = MetadataStore(tmp_path / "contextwiki.sqlite3") + blocking_a = BlockingConnector([document], started, release) + blocking_a.source = SourceAConnector.source + service = IngestionService( + metadata_store=store, + source_registry=SourceRegistry([ + blocking_a, + SourceBConnector([document.model_copy(update={"id": "doc-b", "document_id": "doc-b"})]), + ]), + chunker=DocumentChunker(max_chars=120, overlap_chars=0), + indexer=RecordingIndexer(), + ) + + launched_job = await service.start_sync_source("source_a") + await started.wait() + bulk_task = asyncio.create_task(service.sync_all()) + await asyncio.sleep(0) + assert not bulk_task.done() + release.set() + return launched_job, await bulk_task + + launched_job, result = asyncio.run(run_sync_all_while_background_sync_is_running()) + + blocked = next(item for item in result["results"] if item["source_id"] == "source_a") + succeeded = next(item for item in result["results"] if item["source_id"] == "source_b") + assert launched_job.status == SyncJobStatus.RUNNING + assert result["status"] == "partial" + assert blocked["sync_outcome"] == "blocked" + assert blocked["job"].status == SyncJobStatus.RUNNING + assert succeeded["sync_outcome"] == "succeeded" + assert result["summary"]["blocked"] == 1 + assert result["summary"]["succeeded"] == 1 + + def test_sync_all_reports_failed_when_all_selected_sources_are_blocked(tmp_path): document = DocumentModel( id="doc-a", diff --git a/tests/scripts/test_demo_public_flow.py b/tests/scripts/test_demo_public_flow.py index a0f6d0e..bd50587 100644 --- a/tests/scripts/test_demo_public_flow.py +++ b/tests/scripts/test_demo_public_flow.py @@ -20,8 +20,9 @@ def test_run_demo_returns_grounded_public_flow(): ) ) - assert result["sync"]["status"] == "succeeded" + assert result["sync"]["status"] == "running" assert result["status"]["source"]["source_id"] == "source_obsidian" + assert result["status"]["latest_job"]["status"] == "succeeded" assert result["search"]["results"] assert result["answer"]["evidence_status"] == "grounded" assert result["answer"]["citations"] @@ -35,7 +36,8 @@ def test_run_demo_returns_insufficient_for_unrelated_question(): ) ) - assert result["sync"]["status"] == "succeeded" + assert result["sync"]["status"] == "running" + assert result["status"]["latest_job"]["status"] == "succeeded" assert result["search"]["results"] assert result["answer"]["evidence_status"] == "insufficient" assert ( @@ -104,8 +106,11 @@ def test_render_demo_text_includes_sync_search_and_answer_sections(): text = render_demo_text( { "sample_vault": "sample_vault", - "sync": {"status": "succeeded"}, - "status": {"source": {"source_id": "source_obsidian"}}, + "sync": {"status": "running"}, + "status": { + "source": {"source_id": "source_obsidian"}, + "latest_job": {"status": "succeeded"}, + }, "search": {"results": [{"chunk_id": "chunk-1"}]}, "answer": {"evidence_status": "grounded", "citations": [{"chunk_id": "chunk-1"}]}, }, @@ -130,8 +135,11 @@ def test_render_demo_text_warns_when_search_and_answer_inputs_diverge(): text = render_demo_text( { "sample_vault": "sample_vault", - "sync": {"status": "succeeded"}, - "status": {"source": {"source_id": "source_obsidian"}}, + "sync": {"status": "running"}, + "status": { + "source": {"source_id": "source_obsidian"}, + "latest_job": {"status": "succeeded"}, + }, "search": {"results": [{"chunk_id": "chunk-1"}]}, "answer": {"evidence_status": "grounded", "citations": [{"chunk_id": "chunk-1"}]}, }, @@ -166,10 +174,11 @@ def test_demo_script_json_mode_runs_successfully(): assert payload["query"] == "How does ContextWiki prevent stale citations?" assert payload["question"] == "How does ContextWiki prevent stale citations?" assert payload["same_input"] is True - assert payload["sync"]["status"] == "succeeded" + assert payload["sync"]["status"] == "running" assert payload["sync"]["job_id"] == "" assert payload["status"]["source"]["last_synced_at"] == "" assert payload["status"]["source"]["latest_success_at"] == "" + assert payload["status"]["latest_job"]["status"] == "succeeded" assert payload["search"]["results"][0]["updated_at"] == "" assert payload["answer"]["evidence_status"] == "grounded" assert completed.stdout.lstrip().startswith("{") @@ -242,7 +251,8 @@ def test_demo_public_flow_script_runs_from_repo_root_without_pythonpath(): ) payload = json.loads(completed.stdout) - assert payload["sync"]["status"] == "succeeded" + assert payload["sync"]["status"] == "running" + assert payload["status"]["latest_job"]["status"] == "succeeded" assert payload["answer"]["evidence_status"] == "grounded" assert completed.stdout.lstrip().startswith("{") From 836c4f314c332abd74aa8b901e7a0655d3f01ef6 Mon Sep 17 00:00:00 2001 From: eunhwa99 Date: Mon, 15 Jun 2026 17:43:56 +0900 Subject: [PATCH 2/2] docs: align archived contextwiki notes with main --- .agents/docs/adr/README.md | 27 +- docs/contextwiki-core-understanding.md | 737 ------------------------- 2 files changed, 21 insertions(+), 743 deletions(-) delete mode 100644 docs/contextwiki-core-understanding.md diff --git a/.agents/docs/adr/README.md b/.agents/docs/adr/README.md index 04f6832..1ce51a3 100644 --- a/.agents/docs/adr/README.md +++ b/.agents/docs/adr/README.md @@ -1,12 +1,22 @@ # Architecture Decision Records -This directory records architecture decisions for `MCPContentSearch`. +This directory records historical architecture decisions for +`MCPContentSearch`. -Harness planning reads this index first, then opens only accepted ADRs directly related to the requested change. Review gates treat accepted ADRs as contracts. If a diff conflicts with an accepted ADR, either change the diff or add a new ADR that supersedes the old decision. +For current work, the maintained source of truth is +`.agents/docs/architecture.md`. Harness planning and review do not require +reading ADRs. Keep this directory as optional historical context for older plan +docs, superseded decisions, or cases where a reader explicitly wants the +original decision record. + +If an ADR and `.agents/docs/architecture.md` disagree, follow the architecture +doc and update the historical ADR wording only when that extra archival cleanup +is worth doing. ## ADR Format -Use `template.md` for new records. +Use `template.md` only when you explicitly want to preserve a standalone +decision record in this archive. Required fields: @@ -36,7 +46,10 @@ File names should be numbered and descriptive: ## When to Add or Update ADRs -Add or update an ADR for: +Add or update an ADR only when preserving the separate historical decision +record is itself useful. + +Typical cases: - New cross-module architecture patterns. - MCP tool contract strategy changes. @@ -44,6 +57,8 @@ Add or update an ADR for: - Chroma data migration or reindexing policy. - External integration strategy changes. - Configuration/secrets policy changes. -- Intentional departures from `.agents/docs/architecture.md`. +- Architecture changes worth preserving historically after updating + `.agents/docs/architecture.md`. -Do not add ADRs for ordinary local refactors, one-off bug fixes, or implementation details that do not constrain future work. +Do not add ADRs for ordinary local refactors, one-off bug fixes, or +implementation details that do not constrain future work. diff --git a/docs/contextwiki-core-understanding.md b/docs/contextwiki-core-understanding.md deleted file mode 100644 index ea3c216..0000000 --- a/docs/contextwiki-core-understanding.md +++ /dev/null @@ -1,737 +0,0 @@ -# ContextWiki Core Understanding Note - -Baseline: - -- Original baseline: `eunhwa99/MCPContentSearch` PR #2 -- Current update: slim MCP core refactor with PR #25 Obsidian restoration, - retaining GitHub/Notion/Tistory/Obsidian source sync, SQLite lifecycle - metadata, Chroma retrieval, and citation-backed answers. - -Goal: - -This note is the maintained mental model for explaining ContextWiki's current -design intent, data flow, and limitations. When source connectors, ingestion, -lifecycle metadata, retrieval, citation, or answer behavior changes, update this -note together with README, architecture docs, ADRs, or plan docs as appropriate. - ---- - -## 0. One-line Summary - -ContextWiki is a focused MCP retrieval server. - -The core flow is: - -```text -source registration --> source sync --> document fetch from Notion, Tistory, GitHub, or Obsidian --> external_id / canonical_url / version_id / last_seen metadata normalize --> content_hash computation --> deterministic source-aware chunking for chunk-id comparison --> unchanged documents skip vector reindexing when hash and chunk ids match --> new, changed, reappeared, or rechunked documents are stored in Chroma --> source / job / document / chunk / tombstone metadata is stored in SQLite --> cleanup-capable sources tombstone stale documents after complete snapshots --> search_context may optionally rewrite weak queries through a configured LLM - when CONTEXTWIKI_SEARCH_LLM_ENABLED=true --> answer_with_citations inherits that same query-rewrite egress because it - calls search_context through search_context_for_answer / search_context --> search debug output reports whether rewrite was disabled, skipped, attempted, - or applied --> retrieval starts from Chroma candidates when available, then may add - metadata-fallback candidates before SQLite validation --> retrieval policy keeps vector score and rerank score separate in debug, uses - named rerank/rewrite thresholds in code, and no longer treats a bare - lowercase long token as an implicit GitHub repo probe without explicit - repository-style evidence such as GitHub/repository wording, a strong - repo-shaped anchor, or a curated compound alias --> retrieval candidates are hydrated through SQLite before citation use --> search_context asks Chroma for candidates, may add metadata fallback, and - validates managed hits through SQLite --> search_documents groups those validated candidates by document and keeps one - representative chunk per document for browsing --> answer_with_citations returns a helper evidence-gated answer preview -``` - -Interview or README version: - -```text -ContextWiki exposes MCP tools for source sync, incremental indexing, -citation-ready chunk search, grouped document browsing, context fetch, and a -helper evidence-gated answer preview surface. - -SQLite is the source of truth for lifecycle and citation metadata, while -ChromaDB is the semantic retrieval index. - -In normal production MCP usage, downstream LLMs usually turn retrieved evidence -into the final natural-language answer. `answer_with_citations` is useful here -as a preview/debug/eval helper, not as the repository's core answer contract. -``` - ---- - -## 1. Current Source Coverage - -ContextWiki currently has source connectors for: - -| Source | Source id | How it is configured | Notes | -| --- | --- | --- | --- | -| Notion | `source_notion` | `NOTION_API_KEY` | page/document source | -| Tistory | `source_tistory` | `TISTORY_BLOG_NAME` | blog post source | -| GitHub | `source_github` | `CONTEXTWIKI_GITHUB_REPOSITORIES`, optional `@ref`, `CONTEXTWIKI_GITHUB_DEFAULT_REF`, file limits, optional `GITHUB_TOKEN` | repository file source | -| Obsidian | `source_obsidian` | `CONTEXTWIKI_OBSIDIAN_VAULT_PATH`, `CONTEXTWIKI_OBSIDIAN_MAX_FILES`, `CONTEXTWIKI_OBSIDIAN_MAX_FILE_BYTES` | local Markdown vault source | - -Example: - -```bash -CONTEXTWIKI_GITHUB_REPOSITORIES="eunhwa99/MCPContentSearch@main" -``` - -Then: - -```text -sync_source("source_github") -``` - -fetches supported text/code/Markdown files from configured repositories, -converts each file into a `DocumentModel`, chunks it with line-range citation -metadata, indexes the chunks, and stores lifecycle metadata in SQLite. - -If `@ref` is omitted from a repository spec, the GitHub connector uses -`CONTEXTWIKI_GITHUB_DEFAULT_REF`, which defaults to `main`. Fetch completeness -and stale-cleanup eligibility also depend on `CONTEXTWIKI_GITHUB_MAX_FILES` and -`CONTEXTWIKI_GITHUB_MAX_FILE_BYTES`. `GITHUB_TOKEN` is optional, and -`CONTEXTWIKI_GITHUB_USER_AGENT` controls the request header used by the fetcher. - -Bulk retained-source sync is also available: - -```text -sync_all() -``` - -This fans out one concurrent internal `sync_source()` run per retained source. Each -source still keeps its own SQLite running-job guard, so a source that is -already syncing is reported as blocked in the aggregate result instead of -starting a second overlapping fetch. For the public MCP tool, `sync_source` -now starts or reuses the running job quickly and returns the initial `running` -job payload while the actual ingestion continues in a server-owned background -task. Callers should use `get_sync_status` to observe terminal `succeeded` or -`failed` completion. If the internal blocking ingestion path is cancelled before -completion, the ingestion layer marks that source job failed instead of leaving -a permanent `running` job behind, so a later retry can start cleanly. - -Obsidian example: - -```bash -CONTEXTWIKI_OBSIDIAN_VAULT_PATH="/path/to/temp-or-real-vault" -CONTEXTWIKI_OBSIDIAN_MAX_FILES=2000 -CONTEXTWIKI_OBSIDIAN_MAX_FILE_BYTES=512000 -``` - -Then: - -```text -sync_source("source_obsidian") -``` - -reads bounded `.md` notes from the configured local vault, skips hidden and -Obsidian metadata directories, preserves frontmatter-derived titles when -available, uses `obsidian://open` canonical URLs, and stores lifecycle metadata -in SQLite. If the configured vault exceeds the max file count or per-file byte -bound, sync fails as an incomplete snapshot before stale cleanup. It does not -require a live Obsidian app. - -If a retained connector is disabled for future syncs, already indexed active -documents are not automatically hidden from retrieval. They remain retrievable -until later cleanup or metadata changes mark them inactive. - ---- - -## 2. Overall Mental Model - -ContextWiki is easiest to understand as four layers. - -```mermaid -flowchart TD - Client["AI Client"] - MCP["FastMCP Server"] - Tools["MCP Tools
api/tools.py"] - Ingestion["IngestionService"] - Search["ContextSearchService"] - Answer["CitationAnswerService"] - Store["MetadataStore
SQLite"] - Vector["Vector Index
ChromaDB / LlamaIndex"] - Sources["Source Connectors
Notion / Tistory / GitHub / Obsidian"] - - Client --> MCP - MCP --> Tools - Tools --> Ingestion - Tools --> Search - Tools --> Answer - Ingestion --> Sources - Ingestion --> Store - Ingestion --> Vector - Search --> Vector - Search --> Store - Answer --> Search -``` - -Important design intent: - -```text -AI clients do not read the database directly. -AI clients call MCP tools. - -ChromaDB finds semantically relevant candidate chunks. -SQLite decides whether those chunks are currently active, citeable evidence. -``` - ---- - -## 3. MCP Tool Surface - -Current tools: - -| Tool | Use | -| --- | --- | -| `list_sources()` | see configured sources | -| `sync_source(source_id)` | start one source sync quickly and return the current job | -| `sync_all()` | refresh all retained sources concurrently and return aggregate results | -| `get_sync_status(source_id?)` | inspect source/job state | -| `search_context(query, filters, top_k, include_debug)` | find SQLite-validated evidence | -| `search_documents(query, filters, top_k)` | browse unique matching documents through one representative chunk per document | -| `fetch_context(document_id, chunk_id)` | inspect one document or chunk | -| `answer_with_citations(question, filters, top_k, include_debug)` | helper answer preview from validated evidence | - -The retained `search_context` response always includes a public `debug` key. -On the normal default path `api/tools.py` returns `debug={}` unless -`include_debug=True`, and when caller filters leave no matching public sources -it still returns a small populated `debug` object even if `include_debug=False`. - -The MCP `sync_source(source_id)` tool name and arguments are unchanged, but the -public tool now routes through `IngestionService.start_sync_source()`. A fresh -tool call therefore returns the initial `running` job promptly, or the existing -running job if one is already active. Callers should use -`get_sync_status(source_id)` to observe terminal completion. - -The current debug payload includes rewrite decision fields: - -```text -rewrite_enabled -rewrite_attempted -rewrite_applied -rewrite_skipped_reason -``` - -These fields explain whether query rewrite was disabled, not needed, or tried -without producing an applied rewrite. - -Current rewrite policy keeps one more guardrail: a single high-confidence, -textually matching result may suppress rewrite even when `top_k` asks for more -rows, so `insufficient_candidate_count` alone does not force rewrite churn. - -Observed `rewrite_skipped_reason` values include: - -```text -disabled -not_needed -empty_result -rewrite_failed -no_matching_sources -no_term_groups -not_supported -``` - -Tool handlers call service boundaries and return JSON-safe values through -Pydantic `model_dump(mode="json")` where needed. - -`include_debug=True` is the retained explainability switch for retrieval and -the helper answer preview surface. It is additive and opt-in for populated retrieval reasoning -on the normal success path: default `search_context` payloads still carry -`debug={}`, while debug payloads expose structured retrieval reasoning without -dumping raw local DB contents. The current exception is the public -`no_matching_sources` fast path in `search_context`, which still emits a small -populated `debug` object even when `include_debug=False`. -`answer_with_citations` does not mirror that exception path. - -Verification layer split: - -- Public MCP contract layer: - real `FastMCP.call_tool(...)` payload checks for each retained public tool -- Deterministic functional E2E layer: - retained sync/search/fetch/answer flow modules over temp/local state via - `./scripts/verify_functional_e2e.sh` -- Deterministic quality eval layer: - fixture retrieval and answer scoring via `tests/evals` and - `scripts/run_contextwiki_eval.py` -- Manual live smoke layer: - `scripts/live_query_smoke.py` against a user's configured local runtime; - this is manual diagnostic evidence, not automated gate coverage -Retrieval split: - -```text -search_context -= chunk-level retrieval for evidence, grounding, and citations - -search_documents -= grouped document-browsing retrieval that reuses the same retained-source - candidates but collapses repeated chunks into one representative row per - document -``` - -Helper answer validation focuses on: - -```text -citations -used_chunks -debug -debug_markdown -``` - -Those fields help developers compare the response draft against the retrieved -evidence and decide whether a downstream LLM would have enough grounded context -to answer well. -`list_sources()` and `get_sync_status()` expose additive reviewer-readable -operational fields per source: - -```text -latest_success_at -latest_failure_at -document_count -chunk_count -latest_failure_reason -stale_cleanup_disabled_reason -``` - -Those fields come from SQLite lifecycle metadata plus runtime connector state. -Public error text still passes through the same redaction path used by other -sync/job payloads. - -For reviewer-driven live local verification, the repo also includes: - -```text -scripts/live_query_smoke.py -``` - -This script runs `search_context` and `answer_with_citations` against the local -configured environment through the retained tool handlers and prints a compact -safe summary of rewrite state, hits, helper answer preview status, and -citations. It is a manual live check surface, not a replacement for -deterministic non-live tests. - ---- - -## 4. Core Model Relationships - -Relevant files: - -```text -core/models.py -storage/metadata_store.py -``` - -The most important models are: - -| Model | Meaning | Main use | -| --- | --- | --- | -| `SourceModel` | Notion, Tistory, GitHub, or Obsidian source | source configuration and sync state | -| `SyncJobModel` | one source sync execution | success/failure and processing counts | -| `DocumentModel` | one original document | identity, content hash, lifecycle, source metadata | -| `ChunkModel` | searchable/citeable document segment | vector search and citations | -| `ContextSearchResult` | search response DTO | chunk + score + preview + citation metadata | -| `DocumentSearchResult` | grouped document search response DTO | one representative chunk-backed row per document | - -Key distinction: - -```text -DocumentModel = management and sync unit -Examples: one Notion page, one Tistory post, one GitHub file, one Obsidian note. - -ChunkModel = search and citation unit -Examples: a markdown section, a code line range, a plain-text window. -``` - ---- - -## 5. SQLite vs ChromaDB - -SQLite and Chroma both store chunk-related information, but they have different -jobs. - -```text -SQLite MetadataStore -= source/job/document/chunk lifecycle source of truth - -ChromaDB -= semantic retrieval candidate index -``` - -```mermaid -flowchart TD - A["Fetch from Notion / Tistory / GitHub / Obsidian"] - B["DocumentModel"] - C["DocumentChunker -> ChunkModel[]"] - D["SQLite documents
identity/content/hash/lifecycle"] - E["SQLite chunks
citation metadata and text"] - T["SQLite chunk_tombstones
historical chunk id provenance"] - F["ChromaDB
semantic vector retrieval"] - G["search_context(query)"] - H["Chroma candidate chunks"] - I["Ranking decides whether
metadata fallback is needed"] - J["Combined candidates
Chroma plus metadata fallback"] - N["SQLite active chunk validation"] - O["ContextSearchResult[]"] - K["search_documents(query)"] - L["group by document_id
pick representative chunk"] - M["DocumentSearchResult[]"] - - A --> B --> C - B --> D - C --> E - C --> F - E --> T - G --> H --> I --> J --> N --> O - K --> H - H --> I - I --> J - N --> L --> M -``` - -Search results are not trusted directly from Chroma. Retrieval starts with -vector candidates when available, may add metadata-fallback candidates when the -ranker decides lexical or source-aware recovery is needed, and then hydrates -and validates managed hits through SQLite before they become evidence, -citations, or grouped document rows. - -When `search_context(..., include_debug=True)` is used, the response now makes -that decision path visible with reviewer-readable fields such as: - -```text -intent.name / raw_name / confidence / reasons -query_rewrite.attempted / applied / reason -retrieval_queries -rewritten_queries -filters.source_ids -selected_results[] -``` - -`query_rewrite.reason` is intentionally coarse and stable. Current values are: - -```text -no_initial_candidates -insufficient_candidate_count -missing_textual_match -low_initial_score -``` - -The retrieval path now also makes a coarse intent-policy decision before final -reranking. The current retained intent classes are: - -```text -strict_lookup -broad_topic -list -comparison -``` - -This intent is deterministic and local-first. It is derived from normalized -query terms and anchors, then reused by ranking and grounded answer rendering. -It is not a separate MCP tool and does not require a live LLM call. - ---- - -## 6. Sync and Incremental Indexing - -Relevant files: - -```text -indexing/ingestion_service.py -indexing/chunker.py -indexing/indexer.py -storage/metadata_store.py -``` - -`IngestionService.sync_source()` remains the blocking internal per-source -business flow when it starts the sync itself or joins a same-process local -background task. If another owner already holds the running SQLite job and -there is no local task to join, it returns that current `running` job instead -of blocking on foreign in-flight work. `IngestionService.start_sync_source()` -is the immediate-return background launcher used by the public MCP tool, and -`IngestionService.sync_all()` is the retained-source aggregate fan-out -entrypoint. A just-cancelled same-process background task may be surfaced once -as its reconciled failed job to a direct caller instead of immediately opening -duplicate work, but only while that cancelled job is still the latest -authoritative SQLite job for the source. A successfully completed background -sync does not suppress the next fresh direct `sync_source()` run, and a newer -foreign retry must override any stale local cancelled handoff. - -```text -sync_source(source_id) --> SourceRegistry connector lookup --> MetadataStore register_source + begin_sync_job guard --> connector.fetch_documents() --> normalize document source/id/url/version/last_seen --> compute content_hash --> deterministic source-aware chunking --> skip vector reindexing when active content_hash and chunk ids are unchanged --> index new/changed/reappeared/rechunked chunks in Chroma --> commit document + chunk metadata --> finalize successful sync --> tombstone stale documents when cleanup is safe --> best-effort vector cleanup -``` - -```text -start_sync_source(source_id) --> SourceRegistry connector lookup --> MetadataStore register_source + begin_sync_job guard --> if a job is already running, return that running job --> otherwise spawn the blocking sync_source() work in a background task --> return the initial running job immediately --> caller polls get_sync_status(source_id) for succeeded / failed completion -``` - -```text -sync_all() --> SourceRegistry retained source enumeration --> concurrent per-source sync_source() fan-out --> one aggregate summary with succeeded / failed / blocked counts --> per-source result payloads that preserve the latest job outcome -``` - -Reindexing still happens when: - -```text -- a tombstoned document reappears -- content changes -- generated chunk ids change -- document identity changes -``` - -Failed or partial syncs must not tombstone missing documents. - ---- - -## 7. Source-aware Chunking - -Relevant file: - -```text -indexing/chunker.py -``` - -Current chunking strategy: - -```text -Markdown with headings --> heading / section based chunks - -Markdown without headings --> deterministic plain-text fallback windows - -Code --> deterministic line-range chunks --> blank lines preserved --> long lines split by max_chars - -Plain text --> deterministic character windows -``` - -Each chunk carries citation metadata: - -```text -chunk_id -document_id -source_id -title -url -path -chunk_index -line_start -line_end -content_hash -version_id -updated_at -``` - ---- - -## 8. Document Identity and Versioning - -Relevant fields: - -| Field | Meaning | -| --- | --- | -| `source_id` | which ContextWiki source owns the document | -| `external_id` | stable id from the original system | -| `document_id` | internal canonical document id, usually `external_id` | -| `canonical_url` | primary URL used for citations and legacy matching | -| `version_id` | source version metadata, separate from stable identity | -| `last_seen_at` | last sync time that observed the document | -| `last_seen_sync_id` | job marker that observed the document | -| `deleted_at` | tombstone timestamp when missing from a cleanup-capable successful sync | - -Current mapping: - -```text -Notion --> external_id = page_id --> document_id = page_id - -Tistory --> external_id = blog_name:post_id --> document_id = blog_name:post_id - -GitHub --> external_id/document_id = github:owner/repo:path --> canonical_url = GitHub blob URL at the resolved commit --> version_id = blob SHA - -Obsidian --> external_id/document_id = relative/note/path.md --> canonical_url = obsidian://open URL for the configured vault note --> title = frontmatter title when present, otherwise note stem -``` - -Stable identity should not change just because content changes. `version_id` -records source revision metadata; `content_hash` records actual indexed content. - ---- - -## 9. Tombstones and Stale Vector Safety - -Tombstone means soft delete, not hard delete. - -```text -documents.deleted_at records when a document disappears from a cleanup-capable -source after a complete successful sync. -``` - -For GitHub, cleanup is limited to repository document-id prefixes fetched by the -connector, such as `github:eunhwa99/mcpcontentsearch:`. This keeps one -configured repository sync from tombstoning documents that belong to another -repository scope under the same `source_github` source id. - -For Obsidian, cleanup is allowed only after a complete local-vault snapshot. -Unreadable notes, traversal errors, or exceeded file count/byte bounds should -fail the sync before stale cleanup can tombstone missing active documents. - -Why not hard delete immediately? - -```text -Vector cleanup is best-effort. -If Chroma cleanup fails, SQLite still needs historical provenance to suppress -stale managed vector hits. -``` - -SQLite is the last defense against stale vector results. - ---- - -## 10. Answer Behavior - -`answer_with_citations` delegates retrieval to `ContextSearchService` and only -uses validated chunks as evidence. The answer response should make evidence -status explicit: - -```text -grounded / insufficient / error -``` - -The service must not invent citations from unmanaged or tombstoned chunks. If -retrieval cannot find enough active evidence, the response should say that the -available evidence is insufficient instead of fabricating an answer. - -When evidence is sufficient, the final rendered answer can now vary by request -shape while staying citation-grounded: - -```text -summary-style answer for ordinary grounded QA -Grounded List for collection requests -Grounded Comparison for comparison requests -``` - -This is still bounded synthesis, not open-ended chat. The answer service only -repackages retrieved evidence chunks and keeps the same insufficient-evidence -escape hatch when retrieval is weak. - -`search_documents` does not replace this evidence path. It is a browsing surface -for "which documents matched?" and keeps the highest-ranked chunk as the -representative row so callers can pivot into `fetch_context` or run a later -chunk-level citation workflow through `search_context` or -`answer_with_citations`. - ---- - -## 11. Current Limits - -Current intentional limits: - -- No generic website/docs crawler in production scope. -- No browser Web Console or local HTTP reviewer UI in production scope. -- No Auto Wiki generation or LLM wiki synthesis in production scope. -- No dynamic web fallback or legacy live search/index MCP tools in production - scope. -- No live Obsidian app, plugin, or API server requirement; Obsidian is a - configured local-vault source. -- Optional `CONTEXTWIKI_SEARCH_LLM_ENABLED=true` query rewrite is disabled by - default. If enabled, it may send the user query and normalized terms to the - configured provider, but it must not send source evidence, fetch external - source content, or mutate SQLite/Chroma. -- No deletion, reset, migration, or inspection of local user Chroma/SQLite data - without explicit approval. -- Live Notion, Tistory, GitHub, or embedding-provider validation is opt-in and - approval-gated. Real Obsidian vault validation is also approval-gated; - routine verification uses temporary vaults. - -Historical note: - -- ADR 0004's GitHub connector decision remains current. -- ADR 0004 now also documents the retained Obsidian local-vault connector. -- ADR 0004's website/docs connector portion is superseded for the current scope - by ADR 0006. -- ADR 0005's Auto Wiki decision is superseded for the current scope by ADR - 0006. - ---- - -## 12. Verification Model - -Retained verification layers: - -- Public MCP contract layer: - - `uv run --locked pytest -q tests/contracts/test_public_mcp_contracts.py tests/test_app_composition.py` -- Deterministic functional E2E layer: - - `./scripts/verify_functional_e2e.sh` -- Deterministic quality eval layer: - - `uv run --locked pytest -q tests/evals` - - `uv run --locked python scripts/run_contextwiki_eval.py --output-dir artifacts/contextwiki-evals` - - optional latency artifacts: - `uv run --locked python scripts/run_contextwiki_eval.py --output-dir artifacts/contextwiki-evals --include-latency` -- Manual live smoke layer: - - `uv run --locked python scripts/live_query_smoke.py --query "your query here"` -- Full local gate wrapper: - - `./scripts/verify_all.sh` - - wraps the automated layers above but does not replace the manual live smoke layer - -Functional verification should use fake or temporary persistence and must not -mutate local user Chroma/SQLite data. Live external checks require explicit -approval and should report the source used, safety plan, and whether any local -state was touched. -Obsidian verification should use temporary vault directories unless the user -explicitly approves a real vault path. - -The deterministic eval runner is the retained reviewer-evidence path for -retrieval quality changes. It uses temp SQLite fixture data, never touches user -Chroma/SQLite, and now emits deterministic JSON artifacts with: - -- mixed-query group breakdowns -- retrieval and answer suite pass/fail summaries - -When latency visibility is needed, the runner can also emit a separate -non-deterministic `runtime_metrics.json` file with wall-clock retrieval and -answer timings. That runtime file is an opt-in local artifact from the -latency-enabled command above; the current CI `contextwiki-evals` artifact -uploads only the deterministic retrieval-quality JSON files.