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 242a22b..1ce51a3 100644
--- a/.agents/docs/adr/README.md
+++ b/.agents/docs/adr/README.md
@@ -42,6 +42,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 49b23ea..e890219 100644
--- a/.agents/docs/architecture.md
+++ b/.agents/docs/architecture.md
@@ -113,13 +113,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
 ```
 
 Retained sync safety rule:
diff --git a/README.md b/README.md
index 6491833..e6a7869 100644
--- a/README.md
+++ b/README.md
@@ -25,10 +25,6 @@ vector and metadata stores, then returns verified, citation-backed context.
          [ Verified Context ]
 ```
 
-- **Chroma** handles semantic retrieval.
-- **SQLite** tracks document lifecycle and citation validity.
-- Only chunks still active in SQLite are returned as evidence.
-
 ---
 
 ## MCP Tools
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/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"] == "<generated>"
     assert payload["status"]["source"]["last_synced_at"] == "<generated>"
     assert payload["status"]["source"]["latest_success_at"] == "<generated>"
+    assert payload["status"]["latest_job"]["status"] == "succeeded"
     assert payload["search"]["results"][0]["updated_at"] == "<generated>"
     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("{")