diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index e47f630..8d34096 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -54,7 +54,7 @@ jobs:
         run: python -B 01-capture/tests/test_extension.py
 
   windows-pull:
-    name: Windows public pull, resolution, and fill safety
+    name: Windows pull, research ingest, and context safety
     runs-on: windows-latest
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
@@ -76,6 +76,13 @@ jobs:
           if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
           & prov-sync-unresolved --help | Out-Null
           if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
+          & prov-ingest-research --help | Out-Null
+          if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
+          $contextVersion = & prov-context --version
+          if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
+          if (($contextVersion -join "`n").Trim() -ne '0.18.0') {
+            throw "Unexpected Context Capsule runtime version"
+          }
 
           $raw = & prov-sync-pull --capabilities --json
           $code = $LASTEXITCODE
@@ -98,3 +105,10 @@ jobs:
           }
       - name: Run Windows sync-layer tests
         run: python -m unittest discover -s tests -p "test_synclayer_*.py" -v
+      - name: Run Windows research context tests
+        run: >-
+          python -m unittest
+          tests.test_research_artifacts
+          tests.test_context_capsule
+          tests.test_provenance.TestVaultIngest
+          -v
diff --git a/README.md b/README.md
index 008645a..28e7352 100644
--- a/README.md
+++ b/README.md
@@ -38,6 +38,10 @@ the extension.
 - Organizes conversation records, builds deterministic preparation digests,
   exports a Markdown/Obsidian view, and creates a local SQLite FTS5 index.
 - Exposes read-only research context through a stdlib JSON-RPC/MCP server.
+- Ingests validated `parsed-paper/1.0`, `reading-note/1.0`, `review/1.0`, and
+  `lineage-graph/1.0` artifacts through an atomic, lock-protected local merge.
+- Builds a bounded Context Capsule that separates approved project state,
+  auto-applied low-risk progress, and reference-only literature leads.
 - Captures session-closeout events into a local queue and advances them through a
   resumable, lock-protected state machine.
 - Requires a resolved registered project before a scaffold or
@@ -141,14 +145,22 @@ are unofficial and can break independently of Provenance ingestion.
 
 ```powershell
 uv run --locked prov-ingest-library
+uv run --locked prov-ingest-research C:\ScriptoriumDemo\steward-output
 uv run --locked prov-ingest-vault C:\ScriptoriumDemo\workspace
+uv run --locked prov-context --project synthetic-catalyst --json
 uv run --locked prov-search "research gap"
 ```
 
 The default literature input is
 `%PROVENANCE_HOME%\kb\library.json`. The project portfolio comes from the
 workspace's `Projects/` frontmatter. Obsidian is optional; any Markdown workspace
-that follows the same contract can be used.
+that follows the same contract can be used. Research-artifact inputs must be
+explicit files/directories; the command does not discover personal folders or
+persist their source paths. It rejects symbolic links and Windows reparse points
+in every input or output path component, verifies file identity around reads and
+writes, and redacts absolute local paths including
+`file:` URIs while keeping read/write failures path-free. Capsule literature hints remain reference-only until
+the researcher reviews evidence and approves a claim.
 
 ### Read-only MCP
 
@@ -157,8 +169,8 @@ uv run --locked prov-mcp
 ```
 
 The server exposes `search_brain`, `get_profile`, `list_topics`, `get_portfolio`,
-and `get_current_context` over stdio. It performs no network request and does not
-write project state.
+`get_current_context`, and `get_context_capsule` over stdio. It performs no
+network request and does not write project state.
 
 ### Session writeback
 
@@ -203,6 +215,7 @@ mapping. See [the wiring runbook](docs/sync-layer-wiring.md).
 | `prov-ingest-takeout` | Import a selected Google Takeout ZIP |
 | `prov-ingest-agents` | Import supported local coding-agent logs |
 | `prov-ingest-library` | Import a `library-kb/1.x` snapshot |
+| `prov-ingest-research` | Atomically ingest explicit literature-reading artifacts |
 | `prov-ingest-vault` | Import project frontmatter |
 | `prov-ingest-notes` | Import private Markdown zones into protected state |
 | `prov-scrub` | Re-apply the current privacy transform |
@@ -210,6 +223,7 @@ mapping. See [the wiring runbook](docs/sync-layer-wiring.md).
 | `prov-obsidian` | Export a local Markdown/Obsidian view |
 | `prov-search` | Build/query the local FTS5 index |
 | `prov-mcp` | Start the read-only stdio MCP server |
+| `prov-context` | Print one bounded, read-only project Context Capsule |
 | `prov-sync-pull` | Preview or run one lock-protected sync pass |
 | `prov-sync-pending` / `prov-sync-fill` | Review and submit one safe summary fill |
 | `prov-sync-unresolved` | Inspect unresolved mappings without mutation |
diff --git a/README.zh.md b/README.zh.md
index 37d5f26..c773dc8 100644
--- a/README.zh.md
+++ b/README.zh.md
@@ -31,6 +31,10 @@ Provenance 是 Scriptorium 套件中可独立部署的产品。它通过版本
 - 在记录进入工作层前擦除常见凭据与 PII 形态，并可应用仅保存在本地的别名表。
 - 生成确定性分类、精读输入、Markdown/Obsidian 视图和本地 SQLite FTS5 索引。
 - 通过标准库实现的只读 JSON-RPC/MCP 服务向 Agent 提供研究上下文。
+- 通过带锁的原子合并摄取经过验证的 `parsed-paper/1.0`、
+  `reading-note/1.0`、`review/1.0` 与 `lineage-graph/1.0` 工件。
+- 生成有长度上限的 Context Capsule，明确区分已批准项目状态、自动写入的
+  低风险进度，以及仅供参考的文献线索。
 - 将会话收尾事件写入本地队列，并通过可恢复、带锁的状态机推进。
 - 项目未解析时不生成 scaffold、timeline、draft 或 `session-summary/1.0`。
 - 高价值状态与结论先进入草稿，必须由用户审批后才能写回。
@@ -126,13 +130,20 @@ Provider 内部接口并非公开契约，可能独立于 Provenance 摄取端
 
 ```powershell
 uv run --locked prov-ingest-library
+uv run --locked prov-ingest-research C:\ScriptoriumDemo\steward-output
 uv run --locked prov-ingest-vault C:\ScriptoriumDemo\workspace
+uv run --locked prov-context --project synthetic-catalyst --json
 uv run --locked prov-search "research gap"
 ```
 
 默认文献输入为 `%PROVENANCE_HOME%\kb\library.json`；项目组合来自工作区
 `Projects/` 的 frontmatter。Obsidian 是可选示例，符合相同契约的 Markdown
-工作区也可以使用。
+工作区也可以使用。科研工件必须由用户显式指定文件或目录；命令不会发现个人
+文件夹，也不会保存输入路径；输入和输出路径任一层级中的符号链接、Junction 与
+Windows 重解析点都会被拒绝，读写前后还会核对文件身份，
+包括 `file:` URI 在内的绝对本地路径会被脱敏，读写失败也不会回显本地路径。
+Capsule 中的文献提示在研究者核对证据并批准
+论断前始终只是参考线索。
 
 ### 只读 MCP
 
@@ -141,7 +152,8 @@ uv run --locked prov-mcp
 ```
 
 服务通过 stdio 暴露 `search_brain`、`get_profile`、`list_topics`、
-`get_portfolio` 与 `get_current_context`，不发起网络请求，也不写项目状态。
+`get_portfolio`、`get_current_context` 与 `get_context_capsule`，不发起网络请求，
+也不写项目状态。
 
 ### 会话回写
 
@@ -185,6 +197,7 @@ uv run --locked prov-sync-fill <summary-id> --provenance-home $env:PROVENANCE_HO
 | `prov-ingest-takeout` | 摄取用户选择的 Google Takeout ZIP |
 | `prov-ingest-agents` | 摄取受支持的本地编程 Agent 日志 |
 | `prov-ingest-library` | 摄取 `library-kb/1.x` 快照 |
+| `prov-ingest-research` | 原子摄取显式选择的文献阅读工件 |
 | `prov-ingest-vault` | 摄取项目 frontmatter |
 | `prov-ingest-notes` | 将私人 Markdown 区域摄取进受保护状态 |
 | `prov-scrub` | 对既有记录重跑当前隐私转换 |
@@ -192,6 +205,7 @@ uv run --locked prov-sync-fill <summary-id> --provenance-home $env:PROVENANCE_HO
 | `prov-obsidian` | 导出本地 Markdown/Obsidian 视图 |
 | `prov-search` | 构建或查询本地 FTS5 索引 |
 | `prov-mcp` | 启动只读 stdio MCP 服务 |
+| `prov-context` | 输出一个有边界、只读的项目 Context Capsule |
 | `prov-sync-pull` | 预览或执行一次带锁同步 |
 | `prov-sync-pending` / `prov-sync-fill` | 审阅并提交一个安全摘要 fill |
 | `prov-sync-unresolved` | 只读检查未解析映射 |
diff --git a/docs/OVERVIEW.md b/docs/OVERVIEW.md
index fb332a9..d1cab4c 100644
--- a/docs/OVERVIEW.md
+++ b/docs/OVERVIEW.md
@@ -1,6 +1,6 @@
 # Provenance 产品概览
 
-> 版本 0.17.0 · Scriptorium 套件中的本地科研记忆与证据桥梁。
+> 版本 0.18.0 · Scriptorium 套件中的本地科研记忆与证据桥梁。
 
 ## 1. 用户问题
 
@@ -14,7 +14,7 @@ Provenance 解决的不是单次问答，而是研究项目的连续性：
 - 把支持的本地来源规范化到一个用户控制的数据根；
 - 在进入工作层前执行凭据擦洗与可选假名化；
 - 把文献、项目状态和会话线索接入同一本地检索层；
-- 让 Agent 通过只读 MCP 恢复当前研究上下文；
+- 让 Agent 通过 CLI 或只读 MCP 获得有长度上限、带信任说明的 Context Capsule；
 - 把会话收尾变成可审阅的 timeline、草稿和人工审批，而不是自动改写事实。
 
 ## 2. 产品定位
@@ -42,7 +42,10 @@ Scriptorium 套件内，Steward 负责文献处理，Lectern 负责汇报材料
 脱敏、假名化、规范化、建立本地索引
         |
         v
-Agent 读取当前问题、证据、缺口、决定与下一步
+显式摄取已验证的阅读工件，生成只读 Context Capsule
+        |
+        v
+Agent 读取当前问题、参考证据、缺口、决定与下一步
         |
         v
 一次协作结束 -> 合成 scaffold -> 人工审阅 fill
@@ -67,6 +70,17 @@ Agent 读取当前问题、证据、缺口、决定与下一步
 脱敏降低误泄露概率，不把私人科研内容变成公共数据。附件按原始字节处理，
 尤其不能因文本记录已脱敏就假定附件安全。
 
+Context Capsule 进一步区分三类信息：
+
+| Capsule 内容 | 信任级别 | 使用方式 |
+|---|---|---|
+| 人工编写的项目字段、已批准论断 | `human_or_approved` | 可作为当前项目状态 |
+| 自动追加的低风险进展 | `auto_applied_low_risk_not_approved_claims` | 仅用于恢复过程上下文 |
+| 阅读工件、知识缺口与优先阅读线索 | `reference_only_not_approved_claims` | 只作为待核对参考 |
+
+原始会话、待批草稿、被拒绝论断和无关工件不会进入 Capsule。Markdown 输出
+最多 8,000 字符，JSON 输出最多 20,000 字符；绝对本地路径不会随输出暴露。
+
 ## 5. 首发能力与边界
 
 首发目标是 Windows 上可部署、可验证的本地 Python 产品，要求用户具备配置
@@ -78,6 +92,10 @@ GitHub 开源项目和本地 Agent 的基本能力。Obsidian、Zotero 均为可
 - 本地 Agent 日志、Google Takeout、兼容 ZIP、Markdown、文献库与项目组合摄取；
 - 凭据/PII 擦洗、别名表假名化、分类、精读输入、Obsidian 导出；
 - SQLite FTS5 本地检索与只读 MCP；
+- 显式摄取 `parsed-paper/1.0`、`reading-note/1.0`、`review/1.0` 与
+  `lineage-graph/1.0`，并执行稳定身份、原子幂等合并与搜索索引；
+- 通过 `prov-context` 和 MCP `get_context_capsule` 生成只读、带信任分层的
+  项目恢复上下文；
 - 带锁、可恢复、人工门控的事件/同步层；
 - 手动触发、同源请求、无第三方运行库的可选浏览器导出器；
 - 两个中立 Agent Skill 与跨平台测试。
@@ -87,7 +105,17 @@ GitHub 开源项目和本地 Agent 的基本能力。Obsidian、Zotero 均为可
 - 浏览器商店签名发行，以及对 Provider 未公开内部接口的稳定性保证；
 - 云端托管、后台定时模型调用或自动上传；
 - 未经人工确认的结论自动晋升；
-- 将所有来源内容自动视为事实。
+- 将所有来源内容自动视为事实；
+- `experiment-run/1.0` 与 `claim-evidence/1.0` 的摄取、检索或批准流程。
+
+研究工件必须由用户显式选择文件或目录。摄取器不发现个人目录、不保存来源
+路径，并拒绝符号链接和 Windows 重解析点。包括 `file:` URI 在内的绝对
+Windows、UNC 与 POSIX 路径会在落盘和建立索引前脱敏；读、解析、加锁和写入
+错误不会回显来源或目标路径。
+
+scriptorium-spec 发布实验运行与论断—证据契约只是前置条件，不代表 Provenance
+已经实现相应能力；后续版本仍需单独增加验证、信任与审批语义。0.18.0 会将
+这两种 schema 视为不支持，而不是默认为已有能力。
 
 ## 6. 质量证明
 
@@ -98,11 +126,15 @@ GitHub 开源项目和本地 Agent 的基本能力。Obsidian、Zotero 均为可
   往返和打包 allowlist；
 - 公开树门禁检查敏感文件名、凭据形态、非示例邮箱与已知私有夹具；
 - `skills/` 通过 Skill 结构校验；
-- public pull 默认只预览，写操作需要显式 `--run`。
+- public pull 默认只预览，写操作需要显式 `--run`；
+- 合成测试覆盖研究工件的幂等与并发写入、符号链接/重解析点拒绝、`file:`
+  URI 与路径脱敏、四类工件检索，以及 Capsule 的只读、长度限制和信任分层。
 
 ## 7. 下一阶段
 
-1. 完成浏览器商店发行评估、真实账号兼容性验收与接口变化监测方案。
-2. 提供从合成项目初始化到 MCP 恢复上下文的 Windows 演示路径。
-3. 建立可重复的用户验收：部署时长、首个可检索项目、一次完整 writeback。
+1. 在 scriptorium-spec 契约完成后，为后续版本实现 `experiment-run/1.0`
+   与 `claim-evidence/1.0` 的摄取、信任和审批语义。
+2. 完成浏览器商店发行评估、真实账号兼容性验收与接口变化监测方案。
+3. 建立可重复的 Windows 用户验收：部署时长、首个可检索项目、一次完整
+   Context Capsule 与 writeback。
 4. 为公开契约增加跨仓库兼容测试与版本迁移指南。
diff --git a/docs/SPEC.md b/docs/SPEC.md
index 91b66cc..38f3773 100644
--- a/docs/SPEC.md
+++ b/docs/SPEC.md
@@ -3,7 +3,7 @@
 | Field | Value |
 |---|---|
 | Status | Public living contract |
-| Product version | 0.17.0 |
+| Product version | 0.18.0 |
 | Runtime | Python standard library |
 | Primary release target | Windows 10/11 |
 | Data root | `PROVENANCE_HOME` |
@@ -27,8 +27,9 @@ Included:
 
 - installable `provenance` Python package;
 - local ingestion, privacy transforms, organization, search, Markdown export;
-- literature and project-memory ingestion;
-- read-only stdio MCP server;
+- literature, literature-reading artifact, and project-memory ingestion;
+- bounded project Context Capsules through the CLI and read-only stdio MCP
+  server;
 - event/sync queue, resolver, worker, pending/fill boundary, approvals;
 - optional clean-room browser producer under `01-capture/`;
 - neutral `skills/`, Windows wiring templates, synthetic tests.
@@ -52,7 +53,7 @@ PROVENANCE_HOME/
   inbox/          normalized working records and raw ZIP backup
   drop/           optional user-selected import queue
   kb/             versioned literature input
-  memory/         project, literature and approved-claim snapshots
+  memory/         project, literature, research-artifact and approved-claim snapshots
   output/         deterministic organization views
   _prep/          deterministic Skill inputs
   profile/        local profile artifacts
@@ -108,6 +109,74 @@ through the text redactor.
 RAG/search results and Agent-generated summaries are candidates, not accepted
 research facts.
 
+### 6.1 Research-artifact ingestion
+
+`prov-ingest-research` accepts only user-supplied files and directories. It does
+not discover personal folders, modify source files, or retain source paths. The
+0.18.0 implementation consumes exactly these scriptorium-spec contracts:
+
+- `parsed-paper/1.0`;
+- `reading-note/1.0`;
+- `review/1.0`;
+- `lineage-graph/1.0`.
+
+All selected inputs are validated and sanitized before one deterministic merge
+into `memory/research-artifacts.json`. Records receive stable content identities;
+the load-merge-write transaction is protected by a kernel-backed lock, uses an
+atomic replacement, and is idempotent for unchanged records.
+
+Parsed-paper and reading-note identities preserve the exact Better BibTeX
+citekey. Review and lineage identities include their `created` value, so a later
+snapshot of the same direction is retained while the same direction and
+timestamp remain idempotent.
+
+Input traversal and the destination store boundary fail closed when any path
+component is a symbolic link, junction or Windows reparse point, including
+ancestors and nested entries. Input bytes are read from one verified file
+descriptor and file identity is checked before and after the read. The target
+and kernel-lock identities are rechecked around the load-merge-write transaction.
+Absolute Windows, UNC and POSIX paths and `file:` URIs are redacted before
+persistence and search indexing. Read, parse, lock and write failures return
+bounded messages that do not echo source or destination paths.
+
+### 6.2 Context Capsule
+
+`prov-context --project PROJECT_ID` and MCP `get_context_capsule` return the same
+deterministic, read-only project resume view. The capsule:
+
+- reads human-authored project fields and explicitly approved claims as project
+  state;
+- includes recent auto-applied timeline entries only as low-risk progress, not
+  approved claims;
+- includes related literature-reading artifacts and their gap/priority-reading
+  leads only as reference material, not approved claims;
+- excludes raw conversations, pending drafts, rejected claims, source-path
+  fields and unrelated research artifacts;
+- omits absolute local paths from all returned text;
+- is bounded to 8,000 characters in Markdown and 20,000 characters in the JSON
+  envelope, with explicit truncation metadata.
+
+The trust envelope is part of the public output:
+
+```json
+{
+  "project_state": "human_or_approved",
+  "recent_progress": "auto_applied_low_risk_not_approved_claims",
+  "research_artifacts": "reference_only_not_approved_claims"
+}
+```
+
+Neither the CLI nor the MCP method mutates project or memory state.
+
+### 6.3 Reserved future contracts
+
+Provenance 0.18.0 does not consume `experiment-run/1.0` or
+`claim-evidence/1.0`. Publishing those formats in scriptorium-spec is a
+prerequisite, not an implementation in Provenance: a later Provenance release
+must still add explicit validation, trust semantics and tests. In 0.18.0,
+`prov-ingest-research` rejects them as unsupported schema versions; no
+documentation or Context Capsule output may imply otherwise.
+
 ## 7. Event/sync state machine
 
 ```text
@@ -200,5 +269,5 @@ quick_validate.py skills/session-summary
 ```
 
 Linux and Windows run the extension boundary tests. Linux runs the full Python
-suite; Windows additionally exercises lock, reparse-point, pending/fill and public
-pull behavior.
+suite; Windows additionally exercises lock, reparse-point, research-artifact
+ingestion, Context Capsule, pending/fill and public pull behavior.
diff --git a/provenance/__init__.py b/provenance/__init__.py
index c6ed045..41f8ce9 100644
--- a/provenance/__init__.py
+++ b/provenance/__init__.py
@@ -1,4 +1,4 @@
 """Provenance — local-first capture/sanitize/organize/profile pipeline for your
 cross-AI conversation history. See docs/SPEC.md and docs/PRIVACY.md."""
 
-__version__ = "0.17.0"
+__version__ = "0.18.0"
diff --git a/provenance/context.py b/provenance/context.py
new file mode 100644
index 0000000..1e5bca7
--- /dev/null
+++ b/provenance/context.py
@@ -0,0 +1,518 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Build a bounded, read-only context capsule for one research project.
+
+The capsule is deliberately narrower than ``get_current_context``.  It exposes
+only the human-owned project snapshot, approved high-value overlays, low-risk
+timeline entries, and explicitly labelled reference leads from literature
+artifacts.  It never reads session drafts or raw conversations and it never
+writes to the Provenance data root.
+"""
+import argparse
+import json
+import re
+import sys
+
+from provenance import __version__, memory as M, redact
+
+
+CAPSULE_VERSION = "context-capsule/0.1"
+MAX_MARKDOWN_CHARS = 8000
+MAX_JSON_BYTES = 20000
+# Kept as a compatibility alias for callers that imported the old name.
+MAX_JSON_CHARS = MAX_JSON_BYTES
+
+_MAX_TEXT = 320
+_MAX_ID = 160
+_MAX_ACTIONS = 5
+_MAX_PROGRESS_BLOCKS = 4
+_MAX_PROGRESS_ITEMS = 3
+_MAX_LITERATURE = 6
+_MAX_ARTIFACTS = 8
+_MAX_SOURCE_IDS = 6
+_MAX_REFERENCE_LEADS = 6
+
+_BEGIN = "<!-- scriptorium:progress-log:begin -->"
+_END = "<!-- scriptorium:progress-log:end -->"
+_ENTRY_RE = re.compile(r"<!--\s*entry:[^>]+-->")
+_DATE_RE = re.compile(r"^\*\*([^*]+)\*\*\s*[·-]\s*session writeback\s*$", re.I)
+_FILE_URI_RE = re.compile(
+    r"(?ix)\bfile:(?://)?[^\r\n<>\"'`|)\]]+?"
+    r"(?=\s+https?://|[)\]}>\"'`]|[<|]|[\r\n]|$)"
+)
+_WINDOWS_ABS_RE = re.compile(
+    r"(?i)(?<![A-Za-z0-9/])(?:[A-Z]:[\\/])[^\r\n<>\"'`|]+"
+)
+_UNC_ABS_RE = re.compile(r"(?<![\\])\\\\[^\\/\s]+[\\/][^\r\n<>\"'`|]+")
+_POSIX_ABS_RE = re.compile(
+    r"(?i)(?<![A-Za-z0-9:/])/(?!/)[^\r\n<>\"'`|)\]]+?"
+    r"(?=\s+https?://|[)\]}>\"'`]|[<|]|[\r\n]|$)"
+)
+
+
+class ProjectNotFoundError(LookupError):
+    """Raised when a requested project is absent from the ingested portfolio."""
+
+    def __init__(self, project_id, known_projects=()):
+        self.project_id = project_id
+        self.known_projects = tuple(sorted(str(x) for x in known_projects))
+        super().__init__("Project not found.")
+
+
+def _bounded(value, limit=_MAX_TEXT):
+    """Return one safe, single-line value with local absolute paths omitted."""
+    if value is None:
+        return ""
+    text = str(value)
+    text = _FILE_URI_RE.sub("[local path omitted]", text)
+    text = _WINDOWS_ABS_RE.sub("[local path omitted]", text)
+    text = _UNC_ABS_RE.sub("[local path omitted]", text)
+    text = _POSIX_ABS_RE.sub("[local path omitted]", text)
+    text = redact.redact(text)
+    text = re.sub(r"\s+", " ", text).strip()
+    if len(text) > limit:
+        text = text[: max(0, limit - 1)].rstrip() + "…"
+    return text
+
+
+def _bounded_list(values, count, *, limit=_MAX_TEXT):
+    if not isinstance(values, (list, tuple)):
+        values = [values] if values not in (None, "") else []
+    out = []
+    for value in values:
+        item = _bounded(value, limit)
+        if item and item not in out:
+            out.append(item)
+        if len(out) >= count:
+            break
+    return out
+
+
+def _find_project(project_id, projects):
+    requested = (project_id or "").strip()
+    if not requested:
+        raise ProjectNotFoundError(requested, projects)
+    project = projects.get(requested)
+    if project is None:
+        folded = requested.casefold()
+        project = next(
+            (value for key, value in projects.items() if str(key).casefold() == folded),
+            None,
+        )
+    if not isinstance(project, dict):
+        raise ProjectNotFoundError(requested, projects)
+    return project
+
+
+def _split_human_and_progress(body):
+    text = str(body or "")
+    begin = text.find(_BEGIN)
+    if begin < 0:
+        return text, ""
+    end = text.find(_END, begin + len(_BEGIN))
+    if end < 0:
+        # A malformed generated region is not safe to reinterpret as human prose.
+        return text[:begin], ""
+    human = text[:begin] + text[end + len(_END):]
+    return human, text[begin + len(_BEGIN):end]
+
+
+def _project_goal(project):
+    for key in ("goal", "objective", "research_question", "description", "purpose"):
+        value = _bounded(project.get(key))
+        if value:
+            return value
+
+    human, _ = _split_human_and_progress(project.get("body"))
+    lines = []
+    preferred = []
+    sections = {"question": [], "goal": [], "intuition": []}
+    active_section = None
+    for raw in human.splitlines():
+        line = raw.strip()
+        if not line or line.startswith("<!--"):
+            continue
+        heading = re.match(r"^#{1,6}\s+(.+?)\s*$", line)
+        if heading:
+            heading_text = heading.group(1).strip().casefold()
+            if heading_text in {
+                "research question", "question", "研究问题", "科研问题",
+            }:
+                active_section = "question"
+            elif heading_text in {
+                "research goal", "goal", "objective", "purpose",
+                "研究目标", "科研目标", "项目目标", "目的",
+            }:
+                active_section = "goal"
+            elif heading_text in {
+                "research intuition", "intuition", "研究直觉", "科研直觉",
+            }:
+                active_section = "intuition"
+            else:
+                active_section = None
+            continue
+        if not line:
+            continue
+        match = re.match(
+            r"^(?:research\s+)?(?:goal|objective|question|purpose)\s*[:：]\s*(.+)$",
+            line,
+            re.I,
+        ) or re.match(
+            r"^(?:研究目标|研究问题|科研目标|科研问题|项目目标|目的)\s*[:：]\s*(.+)$",
+            line,
+        )
+        if match:
+            preferred.append(match.group(1))
+        elif active_section:
+            sections[active_section].append(line)
+        else:
+            lines.append(line)
+    section_values = (
+        sections["question"] or sections["goal"] or sections["intuition"]
+    )
+    return _bounded((preferred or section_values or lines or [""])[0])
+
+
+def _recent_progress(body):
+    _, region = _split_human_and_progress(body)
+    if not region:
+        return []
+    out = []
+    for block in _ENTRY_RE.split(region):
+        lines = [line.strip() for line in block.splitlines() if line.strip()]
+        if not lines:
+            continue
+        heading = next((line for line in lines if "session writeback" in line.lower()), "")
+        match = _DATE_RE.match(heading)
+        if not match:
+            continue  # skips approved blocks and unknown generated content
+        items = _bounded_list(
+            [line[2:].strip() for line in lines if line.startswith("- ")],
+            _MAX_PROGRESS_ITEMS,
+        )
+        if items:
+            out.append({"date": _bounded(match.group(1), 40), "items": items})
+        if len(out) >= _MAX_PROGRESS_BLOCKS:
+            break
+    return out
+
+
+def _match_literature(project, library):
+    linked = project.get("linked_literature") or []
+    if not isinstance(linked, list):
+        linked = [linked]
+    direct = {str(value).casefold() for value in linked if str(value).strip()}
+    rows = []
+    for item in library if isinstance(library, list) else []:
+        if not isinstance(item, dict):
+            continue
+        ids = {
+            str(item.get(key) or "").casefold()
+            for key in ("key", "citekey", "doi")
+            if item.get(key)
+        }
+        folders = [str(value) for value in (item.get("folders") or [])]
+        folder_match = any(
+            folder.casefold() == ref or folder.casefold().startswith(ref + "/")
+            for ref in direct
+            for folder in folders
+        )
+        if not (direct.intersection(ids) or folder_match):
+            continue
+        rows.append({
+            "id": _bounded(item.get("key") or item.get("citekey"), _MAX_ID),
+            "citekey": _bounded(item.get("citekey"), _MAX_ID),
+            "title": _bounded(item.get("title")),
+            "year": _bounded(item.get("year"), 20),
+            "read_status": _bounded(item.get("readStatus") or item.get("read_status"), 80),
+            "tldr": _bounded(item.get("tldr")),
+        })
+    rows.sort(key=lambda row: (row["year"], row["title"], row["id"]))
+    return rows
+
+
+def _association_tokens(project, literature):
+    linked = project.get("linked_literature") or []
+    if not isinstance(linked, list):
+        linked = [linked]
+    tokens = {
+        str(value).casefold()
+        for value in linked
+        if str(value).strip()
+    }
+    for item in literature:
+        for key in ("id", "citekey"):
+            if item.get(key):
+                tokens.add(str(item[key]).casefold())
+    return tokens
+
+
+def _raw_artifact_source_ids(kind, content):
+    if kind in ("parsed-paper", "reading-note"):
+        return [content.get("id")]
+    if kind == "review":
+        table = content.get("comparison_table") or {}
+        rows = table.get("rows") if isinstance(table, dict) else []
+        ids = [row.get("citekey") for row in rows if isinstance(row, dict)]
+        ids += list(content.get("priority_reads") or [])
+        return ids
+    if kind == "lineage-graph":
+        return [node.get("citekey") for node in (content.get("nodes") or [])
+                if isinstance(node, dict)]
+    return []
+
+
+def _artifact_source_ids(kind, content):
+    return _bounded_list(
+        _raw_artifact_source_ids(kind, content), _MAX_SOURCE_IDS, limit=_MAX_ID
+    )
+
+
+def _artifact_title(kind, content):
+    if kind == "parsed-paper":
+        metadata = content.get("metadata") or {}
+        return _bounded(metadata.get("title") if isinstance(metadata, dict) else "")
+    if kind == "reading-note":
+        stages = content.get("stages") or {}
+        glance = stages.get("glance") if isinstance(stages, dict) else {}
+        return _bounded(glance.get("tldr") if isinstance(glance, dict) else content.get("id"))
+    direction = content.get("direction") or {}
+    return _bounded(direction.get("query") if isinstance(direction, dict) else "")
+
+
+def _artifact_status(kind, content):
+    if kind == "parsed-paper":
+        return _bounded(content.get("parser") or "parsed", 80)
+    if kind == "reading-note":
+        return _bounded(content.get("read_status") or "reading note", 80)
+    if kind == "review":
+        return "direction synthesis"
+    if kind == "lineage-graph":
+        return "lineage graph"
+    return "reference"
+
+
+def _artifact_summary(kind, content):
+    if kind == "parsed-paper":
+        return _bounded(
+            "Structured parse available: "
+            f"{len(content.get('sections') or [])} section(s), "
+            f"{len(content.get('references') or [])} reference(s), "
+            f"{len(content.get('figures') or [])} figure(s)."
+        )
+    if kind == "reading-note":
+        stages = content.get("stages") or {}
+        names = sorted(stages) if isinstance(stages, dict) else []
+        glance = stages.get("glance") if isinstance(stages, dict) else {}
+        tldr = glance.get("tldr") if isinstance(glance, dict) else ""
+        return _bounded(tldr or ("Reading stages available: " + ", ".join(names)))
+    if kind == "review":
+        table = content.get("comparison_table") or {}
+        rows = table.get("rows") if isinstance(table, dict) else []
+        return _bounded(
+            f"Direction synthesis available: {len(content.get('sections') or [])} section(s), "
+            f"{len(rows or [])} compared paper(s)."
+        )
+    if kind == "lineage-graph":
+        return _bounded(
+            f"Lineage source available: {len(content.get('nodes') or [])} paper(s), "
+            f"{len(content.get('edges') or [])} relation(s)."
+        )
+    return "Reference artifact available."
+
+
+def _load_related_artifacts(project, literature, root):
+    loader = getattr(M, "load_research_artifacts", None)
+    loaded = loader(root) if callable(loader) else []
+    tokens = _association_tokens(project, literature)
+    rows = []
+    gaps = []
+    priority_reads = []
+    for record in loaded if isinstance(loaded, list) else []:
+        if not isinstance(record, dict):
+            continue
+        kind = record.get("artifact_type")
+        content = record.get("content")
+        if kind not in {"parsed-paper", "reading-note", "review", "lineage-graph"}:
+            continue
+        if not isinstance(content, dict):
+            continue
+        raw_source_ids = _raw_artifact_source_ids(kind, content)
+        if not tokens.intersection(
+                str(value).casefold() for value in raw_source_ids if value):
+            continue
+        source_ids = _bounded_list(raw_source_ids, _MAX_SOURCE_IDS, limit=_MAX_ID)
+        direction = content.get("direction") or {}
+        query = direction.get("query") if isinstance(direction, dict) else ""
+        artifact_id = record.get("artifact_id") or content.get("id") or query
+        rows.append({
+            "kind": kind,
+            "schema_version": _bounded(record.get("schema_version") or content.get("schema_version"), 80),
+            "id": _bounded(artifact_id, _MAX_ID),
+            "title": _artifact_title(kind, content),
+            "status": _artifact_status(kind, content),
+            "source_ids": source_ids,
+            "summary": _artifact_summary(kind, content),
+            "trust": "reference_only",
+        })
+        if kind == "review":
+            gaps.extend(content.get("gaps") or [])
+            priority_reads.extend(content.get("priority_reads") or [])
+
+    rows.sort(key=lambda row: (row["kind"], row["id"], row["title"]))
+    return (
+        rows[:_MAX_ARTIFACTS],
+        _bounded_list(gaps, _MAX_REFERENCE_LEADS),
+        _bounded_list(priority_reads, _MAX_REFERENCE_LEADS, limit=_MAX_ID),
+    )
+
+
+def _render_unbounded(capsule):
+    p = capsule["project"]
+    out = [
+        f"# Context Capsule: {p['title'] or p['project_id']}",
+        "",
+        "> Trust boundary: project fields are human-authored or explicitly approved. "
+        "Recent progress is auto-applied low-risk context, not an approved claim. "
+        "Literature artifacts and leads below are references/candidates, not approved claims.",
+        "",
+        "## Project state",
+        f"- Project ID: {p['project_id']}",
+        f"- Status: {p['status'] or '?'}",
+    ]
+    for label, key in (("Stage", "stage"), ("Priority", "priority"), ("Updated", "updated"),
+                       ("Goal", "goal"), ("Approved conclusion", "conclusion")):
+        if p[key]:
+            out.append(f"- {label}: {p[key]}")
+
+    if capsule["next_actions"] or capsule["blocked_by"]:
+        out += ["", "## Gaps and next steps"]
+        for action in capsule["next_actions"]:
+            out.append(f"- Next: {action}")
+        if capsule["blocked_by"]:
+            out.append(f"- Blocked by: {capsule['blocked_by']}")
+
+    if capsule["recent_progress"]:
+        out += ["", "## Recent low-risk progress"]
+        for block in capsule["recent_progress"]:
+            for item in block["items"]:
+                out.append(f"- {block['date']}: {item}")
+
+    if capsule["literature"]:
+        out += ["", "## Linked literature"]
+        for item in capsule["literature"]:
+            suffix = " · ".join(x for x in (item["year"], item["read_status"]) if x)
+            out.append(f"- {item['title'] or item['citekey'] or item['id']}"
+                       + (f" ({suffix})" if suffix else ""))
+            if item["tldr"]:
+                out.append(f"  Reference summary: {item['tldr']}")
+
+    if capsule["research_artifacts"]:
+        out += ["", "## Research artifact hints (reference only)"]
+        for item in capsule["research_artifacts"]:
+            out.append(f"- [{item['kind']}] {item['title'] or item['id']}: {item['summary']}")
+
+    leads = capsule["reference_leads"]
+    if leads["gaps"] or leads["priority_reads"]:
+        out += ["", "## Candidate leads from literature (not approved claims)"]
+        for gap in leads["gaps"]:
+            out.append(f"- Candidate gap: {gap}")
+        for citekey in leads["priority_reads"]:
+            out.append(f"- Suggested reading: {citekey}")
+    return "\n".join(out).rstrip() + "\n"
+
+
+def render_markdown(capsule):
+    """Render a deterministic Markdown view, always within the hard limit."""
+    text = _render_unbounded(capsule)
+    if len(text) <= MAX_MARKDOWN_CHARS:
+        return text
+    marker = "\n\n[Capsule truncated at the 8000-character safety limit]\n"
+    return text[: MAX_MARKDOWN_CHARS - len(marker)].rstrip() + marker
+
+
+def _fit_json(capsule):
+    """Keep the JSON CLI output bounded without changing its public envelope."""
+    serialized_size = lambda: len(
+        json.dumps(capsule, ensure_ascii=False, indent=2).encode("utf-8")
+    )
+    for key in ("research_artifacts", "literature", "recent_progress"):
+        while capsule[key] and serialized_size() > MAX_JSON_BYTES:
+            capsule[key].pop()
+            capsule["limits"]["truncated"] = True
+    while capsule["reference_leads"]["gaps"] and serialized_size() > MAX_JSON_BYTES:
+        capsule["reference_leads"]["gaps"].pop()
+        capsule["limits"]["truncated"] = True
+    while (
+        capsule["reference_leads"]["priority_reads"]
+        and serialized_size() > MAX_JSON_BYTES
+    ):
+        capsule["reference_leads"]["priority_reads"].pop()
+        capsule["limits"]["truncated"] = True
+    return capsule
+
+
+def build_capsule(project_id, root=None):
+    """Return a deterministic, bounded, read-only context capsule dictionary."""
+    project = _find_project(project_id, M.load_projects(root))
+    matched_literature = _match_literature(project, M.load_library(root))
+    artifacts, gaps, priority_reads = _load_related_artifacts(
+        project, matched_literature, root
+    )
+    literature = matched_literature[:_MAX_LITERATURE]
+    canonical_id = _bounded(project.get("project_id") or project_id, _MAX_ID)
+    capsule = {
+        "capsule_version": CAPSULE_VERSION,
+        "project": {
+            "project_id": canonical_id,
+            "title": _bounded(project.get("title")),
+            "status": _bounded(project.get("status"), 80),
+            "stage": _bounded(project.get("stage")),
+            "priority": _bounded(project.get("priority"), 80),
+            "updated": _bounded(project.get("updated"), 80),
+            "goal": _project_goal(project),
+            "conclusion": _bounded(project.get("conclusion")),
+        },
+        "next_actions": _bounded_list(project.get("next_actions"), _MAX_ACTIONS),
+        "blocked_by": _bounded(project.get("blocked_by")),
+        "recent_progress": _recent_progress(
+            project.get("_progress_log") or project.get("body")
+        ),
+        "literature": literature,
+        "research_artifacts": artifacts,
+        "reference_leads": {"gaps": gaps, "priority_reads": priority_reads},
+        "trust": {
+            "project_state": "human_or_approved",
+            "recent_progress": "auto_applied_low_risk_not_approved_claims",
+            "research_artifacts": "reference_only_not_approved_claims",
+        },
+        "limits": {"max_markdown_chars": MAX_MARKDOWN_CHARS, "truncated": False},
+    }
+    _fit_json(capsule)
+    if len(_render_unbounded(capsule)) > MAX_MARKDOWN_CHARS:
+        capsule["limits"]["truncated"] = True
+    return capsule
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(
+        description="Print a bounded, read-only research-project context capsule."
+    )
+    parser.add_argument("--version", action="version", version=__version__)
+    parser.add_argument("--project", required=True, help="Stable project ID")
+    parser.add_argument("--json", action="store_true", help="Emit the structured JSON envelope")
+    args = parser.parse_args(argv)
+    try:
+        capsule = build_capsule(args.project)
+    except ProjectNotFoundError as exc:
+        print(str(exc), file=sys.stderr)
+        return 2
+    if args.json:
+        print(json.dumps(capsule, ensure_ascii=False, indent=2))
+    else:
+        print(render_markdown(capsule), end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/provenance/ingest/research.py b/provenance/ingest/research.py
new file mode 100644
index 0000000..3b59abc
--- /dev/null
+++ b/provenance/ingest/research.py
@@ -0,0 +1,706 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Ingest Scriptorium literature-research artifacts into Provenance memory.
+
+The command accepts only explicit files/directories. It never discovers user
+data, modifies sources, or persists source paths. Every input is validated and
+sanitized before one atomic, idempotent merge into
+``memory/research-artifacts.json``.
+
+Supported contracts:
+    parsed-paper/1.0
+    reading-note/1.0
+    review/1.0
+    lineage-graph/1.0
+
+Usage:
+    prov-ingest-research ARTIFACT.json [ARTIFACT.json ...]
+    prov-ingest-research OUTPUT_DIR [OUTPUT_DIR ...]
+    prov-ingest-research INPUT [INPUT ...] --memory-dir MEMORY_DIR
+"""
+import argparse
+import hashlib
+import json
+import os
+import re
+import stat
+import sys
+import tempfile
+import unicodedata
+from pathlib import Path
+
+from provenance import common as C, memory as M
+from provenance.redact import sanitize_json
+from provenance.synclayer.worker import (
+    WorkerBusy,
+    WorkerStateError,
+    _Lock as _KernelLock,
+)
+
+
+SUPPORTED = {
+    "parsed-paper/1.0": "parsed-paper",
+    "reading-note/1.0": "reading-note",
+    "review/1.0": "review",
+    "lineage-graph/1.0": "lineage-graph",
+}
+STORE_VERSION = "research-artifacts/1.0"
+RELATIONS = {"cites", "extends", "supersedes", "method-of", "contrasts"}
+READ_STATUSES = {"New", "To Read", "In Progress", "Read", "Not Reading"}
+
+_WINDOWS_ABSOLUTE = re.compile(
+    r"(?i)(?<![A-Za-z0-9/])[A-Z]:[\\/][^\r\n<>\"'`|]+"
+)
+_UNC_ABSOLUTE = re.compile(r"(?<![\\])\\\\[^\\/\s]+[\\/][^\r\n<>\"'`|]+")
+_POSIX_ABSOLUTE = re.compile(r"(?<![A-Za-z0-9:/])/(?!/)[^\r\n<>\"'`|]+")
+_FILE_URI = re.compile(
+    r"(?ix)\bfile:(?://)?[^\r\n<>\"'`|)\]]+?"
+    r"(?=\s+https?://|[)\]}>\"'`]|[<|]|[\r\n]|$)"
+)
+
+
+class ArtifactError(ValueError):
+    """A validation/read error safe to show without an absolute source path."""
+
+
+def _nonempty_string(value, field):
+    if not isinstance(value, str) or not value.strip():
+        raise ArtifactError(f"{field} must be a non-empty string")
+
+
+def _optional_string(obj, field, where):
+    if field in obj and not isinstance(obj[field], str):
+        raise ArtifactError(f"{where}.{field} must be a string")
+
+
+def _string_list(value, field):
+    if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
+        raise ArtifactError(f"{field} must be an array of strings")
+
+
+def _object(value, field):
+    if not isinstance(value, dict):
+        raise ArtifactError(f"{field} must be an object")
+    return value
+
+
+def _object_list(value, field):
+    if not isinstance(value, list) or any(not isinstance(item, dict) for item in value):
+        raise ArtifactError(f"{field} must be an array of objects")
+    return value
+
+
+def _validate_parsed_paper(doc):
+    _nonempty_string(doc.get("id"), "id")
+    _nonempty_string(doc.get("created"), "created")
+    for field in ("parser", "source_pdf"):
+        _optional_string(doc, field, "parsed-paper")
+    if "metadata" in doc:
+        metadata = _object(doc["metadata"], "metadata")
+        for field in ("title", "year", "doi", "abstract"):
+            _optional_string(metadata, field, "metadata")
+        if "authors" in metadata:
+            _string_list(metadata["authors"], "metadata.authors")
+    for field in ("sections", "references", "figures", "tables"):
+        if field not in doc:
+            continue
+        items = _object_list(doc[field], field)
+        for index, item in enumerate(items):
+            where = f"{field}[{index}]"
+            if field == "sections":
+                for key in ("heading", "text"):
+                    _optional_string(item, key, where)
+                if "level" in item and type(item["level"]) is not int:
+                    raise ArtifactError(f"{where}.level must be an integer")
+            elif field == "references":
+                for key in ("raw", "title", "year", "doi"):
+                    _optional_string(item, key, where)
+                if "authors" in item:
+                    _string_list(item["authors"], f"{where}.authors")
+            elif field == "figures":
+                for key in ("id", "caption"):
+                    _optional_string(item, key, where)
+                if "page" in item and type(item["page"]) is not int:
+                    raise ArtifactError(f"{where}.page must be an integer")
+            else:
+                for key in ("id", "caption"):
+                    _optional_string(item, key, where)
+                if "n_rows" in item and type(item["n_rows"]) is not int:
+                    raise ArtifactError(f"{where}.n_rows must be an integer")
+
+
+def _validate_reading_note(doc):
+    _nonempty_string(doc.get("id"), "id")
+    _nonempty_string(doc.get("created"), "created")
+    for field in ("doi", "generated_by"):
+        _optional_string(doc, field, "reading-note")
+    if "zotero_key" in doc:
+        value = doc["zotero_key"]
+        if not isinstance(value, str) or re.fullmatch(r"[A-Z0-9]{8}", value) is None:
+            raise ArtifactError("zotero_key must contain exactly eight uppercase letters/digits")
+    if "read_status" in doc:
+        if not isinstance(doc["read_status"], str) or doc["read_status"] not in READ_STATUSES:
+            raise ArtifactError("read_status is not a supported reading state")
+    if "stages" in doc:
+        stages = _object(doc["stages"], "stages")
+        stage_fields = {
+            "glance": {"tldr": "string", "tags": "strings", "key_findings": "strings"},
+            "close_read": {"question": "string", "method": "string", "data": "string",
+                           "results": "string", "figures": "strings"},
+            "deep_read": {"critique": "string", "reproducibility": "string",
+                          "limits": "string", "relation_to_my_work": "string"},
+            "situate": {"direction": "string", "lineage_refs": "strings"},
+        }
+        for stage, fields in stage_fields.items():
+            if stage not in stages:
+                continue
+            value = _object(stages[stage], f"stages.{stage}")
+            for field, expected in fields.items():
+                if field not in value:
+                    continue
+                if expected == "string":
+                    _optional_string(value, field, f"stages.{stage}")
+                else:
+                    _string_list(value[field], f"stages.{stage}.{field}")
+    if "sources" in doc:
+        sources = _object(doc["sources"], "sources")
+        if "annotations" in sources:
+            _string_list(sources["annotations"], "sources.annotations")
+        for field in ("parsed_file", "zotero_uri"):
+            _optional_string(sources, field, "sources")
+
+
+def _validate_review(doc):
+    direction = _object(doc.get("direction"), "direction")
+    _nonempty_string(direction.get("query"), "direction.query")
+    _nonempty_string(direction.get("created"), "direction.created")
+    sections = _object_list(doc.get("sections"), "sections")
+    for index, section in enumerate(sections):
+        _nonempty_string(section.get("heading"), f"sections[{index}].heading")
+        if not isinstance(section.get("prose"), str):
+            raise ArtifactError(f"sections[{index}].prose must be a string")
+    if "comparison_table" in doc:
+        table = _object(doc["comparison_table"], "comparison_table")
+        if "columns" in table:
+            _string_list(table["columns"], "comparison_table.columns")
+        if "rows" in table:
+            rows = _object_list(table["rows"], "comparison_table.rows")
+            for index, row in enumerate(rows):
+                _nonempty_string(row.get("citekey"), f"comparison_table.rows[{index}].citekey")
+                _string_list(row.get("cells"), f"comparison_table.rows[{index}].cells")
+    for field in ("gaps", "priority_reads"):
+        if field in doc:
+            _string_list(doc[field], field)
+    _optional_string(doc, "lineage_ref", "review")
+
+
+def _validate_lineage(doc):
+    direction = _object(doc.get("direction"), "direction")
+    for field in ("query", "scope_method", "created"):
+        if not isinstance(direction.get(field), str):
+            raise ArtifactError(f"direction.{field} must be a string")
+    _nonempty_string(direction.get("created"), "direction.created")
+    nodes = _object_list(doc.get("nodes"), "nodes")
+    for index, node in enumerate(nodes):
+        _nonempty_string(node.get("citekey"), f"nodes[{index}].citekey")
+        for field in ("title", "year", "cluster"):
+            _optional_string(node, field, f"nodes[{index}]")
+    edges = _object_list(doc.get("edges"), "edges")
+    for index, edge in enumerate(edges):
+        for field in ("from", "to", "relation"):
+            _nonempty_string(edge.get(field), f"edges[{index}].{field}")
+        if edge["relation"] not in RELATIONS:
+            raise ArtifactError(f"edges[{index}].relation is not supported")
+        _optional_string(edge, "evidence", f"edges[{index}]")
+    if "clusters" in doc:
+        clusters = _object_list(doc["clusters"], "clusters")
+        for index, cluster in enumerate(clusters):
+            _nonempty_string(cluster.get("id"), f"clusters[{index}].id")
+            _nonempty_string(cluster.get("label"), f"clusters[{index}].label")
+    if "timeline" in doc:
+        _string_list(doc["timeline"], "timeline")
+    _optional_string(doc, "generated_by", "lineage-graph")
+
+
+VALIDATORS = {
+    "parsed-paper": _validate_parsed_paper,
+    "reading-note": _validate_reading_note,
+    "review": _validate_review,
+    "lineage-graph": _validate_lineage,
+}
+
+
+def validate_artifact(doc):
+    """Validate the strict minimum needed for safe identity and search indexing."""
+    if not isinstance(doc, dict):
+        raise ArtifactError("artifact must be a JSON object")
+    version = doc.get("schema_version")
+    artifact_type = SUPPORTED.get(version)
+    if artifact_type is None:
+        raise ArtifactError("unsupported schema_version")
+    VALIDATORS[artifact_type](doc)
+    return artifact_type
+
+
+def _redact_local_paths(value):
+    """Remove absolute local paths while retaining safe relative contract refs."""
+    if isinstance(value, str):
+        stripped = value.strip()
+        if (re.match(r"(?i)^file:", stripped)
+                or re.match(r"(?i)^[A-Z]:[\\/]", stripped) or stripped.startswith("\\\\")
+                or stripped.startswith("/")):
+            return "[REDACTED:local-path]"
+        value = _FILE_URI.sub("[REDACTED:local-path]", value)
+        value = _WINDOWS_ABSOLUTE.sub("[REDACTED:local-path]", value)
+        value = _UNC_ABSOLUTE.sub("[REDACTED:local-path]", value)
+        return _POSIX_ABSOLUTE.sub("[REDACTED:local-path]", value)
+    if isinstance(value, list):
+        return [_redact_local_paths(item) for item in value]
+    if isinstance(value, dict):
+        clean = {}
+        for key, item in value.items():
+            safe_key = (
+                sanitize_json(_redact_local_paths(key))
+                if isinstance(key, str)
+                else key
+            )
+            if safe_key in clean:
+                base = str(safe_key)
+                index = 2
+                while f"{base}#{index}" in clean:
+                    index += 1
+                safe_key = f"{base}#{index}"
+            clean[safe_key] = _redact_local_paths(item)
+        return clean
+    return value
+
+
+def _identity(artifact_type, doc):
+    if artifact_type in ("parsed-paper", "reading-note"):
+        return doc["id"]
+    direction = doc["direction"]
+    query = doc["direction"]["query"]
+    if query.strip() or artifact_type == "review":
+        base = query
+    else:
+        # The lineage contract explicitly permits an empty ad-hoc query. The paper
+        # set is its durable semantic identity in that case; input filenames are
+        # intentionally never used because they may move or contain private paths.
+        citekeys = sorted(node["citekey"] for node in doc["nodes"])
+        base = "\0".join(citekeys) if citekeys else direction["scope_method"]
+    return f"{base}\0{direction['created']}"
+
+
+def _artifact_id(artifact_type, doc):
+    identity = unicodedata.normalize("NFKC", _identity(artifact_type, doc)).strip()
+    query = str(doc.get("direction", {}).get("query", ""))
+    if (
+        artifact_type not in ("parsed-paper", "reading-note")
+        and (artifact_type == "review" or query.strip())
+    ):
+        identity = identity.casefold()
+    digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:24]
+    return f"{artifact_type}:{digest}"
+
+
+def normalize_artifact(doc):
+    """Return one stable, path-free and privacy-sanitized memory record."""
+    artifact_type = validate_artifact(doc)
+    clean = sanitize_json(_redact_local_paths(doc))
+    return {
+        "artifact_id": _artifact_id(artifact_type, doc),
+        "artifact_type": artifact_type,
+        "schema_version": doc["schema_version"],
+        "content": clean,
+    }
+
+
+def _atomic_write_json(path, document):
+    path = Path(path)
+    _prepare_safe_directory(path.parent)
+    _optional_safe_file(path)
+    fd, temp_name = tempfile.mkstemp(
+        prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent)
+    )
+    temp = Path(temp_name)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
+            json.dump(document, handle, ensure_ascii=False, indent=2, sort_keys=True)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        _require_safe_directory(path.parent)
+        _optional_safe_file(path)
+        os.replace(temp, path)
+        _require_safe_directory(path.parent)
+        _, stored = _optional_safe_file(path)
+        if stored is None:
+            raise ArtifactError("the research artifact store was not published safely")
+    except Exception:
+        try:
+            _, temp_info = _optional_safe_file(temp)
+            if temp_info is not None:
+                temp.unlink()
+        except (ArtifactError, OSError):
+            pass
+        raise
+
+
+def _load_existing(path):
+    path = Path(path)
+    try:
+        _, info = _optional_safe_file(path)
+    except ArtifactError as exc:
+        raise ArtifactError("existing memory store is unreadable") from exc
+    if info is None:
+        return []
+    try:
+        document = json.loads(_read_safe_bytes(path, expected=_file_snapshot(info)).decode("utf-8"))
+    except Exception as exc:
+        raise ArtifactError("existing memory store is unreadable") from exc
+    if not isinstance(document, dict) or document.get("schema_version") != STORE_VERSION:
+        raise ArtifactError("existing memory store has an unsupported schema")
+    artifacts = document.get("artifacts")
+    if not isinstance(artifacts, list):
+        raise ArtifactError("existing memory store artifacts must be an array")
+    seen = set()
+    for index, record in enumerate(artifacts):
+        if not isinstance(record, dict):
+            raise ArtifactError(f"existing artifacts[{index}] must be an object")
+        artifact_id = record.get("artifact_id")
+        if not isinstance(artifact_id, str) or not artifact_id or artifact_id in seen:
+            raise ArtifactError("existing memory store has invalid or duplicate artifact IDs")
+        if record.get("artifact_type") not in VALIDATORS or not isinstance(record.get("content"), dict):
+            raise ArtifactError(f"existing artifacts[{index}] has an invalid record shape")
+        seen.add(artifact_id)
+    return artifacts
+
+
+class _ValidatedStoreLock:
+    """Kernel lock whose path and fd remain bound to the selected safe store."""
+
+    def __init__(self, target):
+        self.target = Path(target)
+        self.lock_path = self.target.with_name(f".{self.target.name}.lock")
+        self.kernel_lock = _KernelLock(self.lock_path)
+        self.identity = None
+
+    def _validate_open_lock(self):
+        _require_safe_directory(self.lock_path.parent)
+        _, current = _optional_safe_file(self.lock_path)
+        if current is None or self.kernel_lock.fd is None:
+            raise WorkerStateError("research store lock identity is unavailable")
+        try:
+            opened = os.fstat(self.kernel_lock.fd)
+        except OSError as exc:
+            raise WorkerStateError("research store lock identity is unavailable") from exc
+        if _file_identity(current) != _file_identity(opened):
+            raise WorkerStateError("research store lock identity changed")
+        if self.identity is not None and _file_identity(current) != self.identity:
+            raise WorkerStateError("research store lock identity changed")
+        self.identity = _file_identity(current)
+
+    def validate(self):
+        try:
+            self._validate_open_lock()
+            _reject_unsafe_existing(self.target)
+        except ArtifactError as exc:
+            raise WorkerStateError("research store filesystem boundary changed") from exc
+
+    def __enter__(self):
+        try:
+            _prepare_safe_directory(self.lock_path.parent)
+            _reject_unsafe_existing(self.target)
+            _optional_safe_file(self.lock_path)
+        except ArtifactError as exc:
+            raise WorkerStateError("research store filesystem boundary is unsafe") from exc
+        self.kernel_lock.__enter__()
+        try:
+            self.validate()
+        except Exception:
+            self.kernel_lock.__exit__(*sys.exc_info())
+            raise
+        return self
+
+    def __exit__(self, *exc_info):
+        validation_error = None
+        try:
+            self.validate()
+        except (ArtifactError, WorkerStateError) as exc:
+            validation_error = exc
+        result = self.kernel_lock.__exit__(*exc_info)
+        if validation_error is not None and exc_info[1] is None:
+            raise WorkerStateError("research store filesystem boundary changed") from validation_error
+        return result
+
+
+def _store_lock(target):
+    """Serialize read-merge-write and bind the lock to the safe store path."""
+    return _ValidatedStoreLock(target)
+
+
+def _path_key(path):
+    return os.path.normcase(os.path.abspath(os.fspath(path)))
+
+
+def _is_link_or_reparse(info):
+    reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
+    attributes = getattr(info, "st_file_attributes", 0)
+    return stat.S_ISLNK(info.st_mode) or bool(reparse_flag and attributes & reparse_flag)
+
+
+def _path_is_link_or_reparse(path, info):
+    if _is_link_or_reparse(info):
+        return True
+    is_junction = getattr(path, "is_junction", None)
+    try:
+        return bool(is_junction and is_junction())
+    except OSError as exc:
+        raise ArtifactError("a filesystem boundary could not be verified safely") from exc
+
+
+def _absolute_components(path):
+    absolute = Path(os.path.abspath(os.fspath(path)))
+    anchor = Path(absolute.anchor)
+    components = [anchor]
+    current = anchor
+    for part in absolute.parts[1:]:
+        current = current / part
+        components.append(current)
+    return absolute, components
+
+
+def _safe_chain(path, *, allow_missing_leaf=False, allow_missing_tail=False):
+    """Return ``(absolute_path, final_lstat)`` after checking every component."""
+    absolute, components = _absolute_components(path)
+    final = None
+    last_index = len(components) - 1
+    for index, component in enumerate(components):
+        try:
+            info = component.lstat()
+        except FileNotFoundError:
+            if allow_missing_tail or (allow_missing_leaf and index == last_index):
+                return absolute, None
+            raise ArtifactError("a filesystem boundary is unavailable")
+        except OSError as exc:
+            raise ArtifactError("a filesystem boundary could not be verified safely") from exc
+        if _path_is_link_or_reparse(component, info):
+            raise ArtifactError("filesystem boundaries cannot contain links or reparse points")
+        final = info
+    return absolute, final
+
+
+def _file_identity(info):
+    return info.st_dev, info.st_ino
+
+
+def _file_snapshot(info):
+    return (
+        info.st_dev,
+        info.st_ino,
+        info.st_size,
+        getattr(info, "st_mtime_ns", int(info.st_mtime * 1_000_000_000)),
+    )
+
+
+def _require_safe_directory(path):
+    absolute, info = _safe_chain(path)
+    if info is None or not stat.S_ISDIR(info.st_mode):
+        raise ArtifactError("a filesystem boundary is not a directory")
+    return absolute, info
+
+
+def _prepare_safe_directory(path):
+    absolute, _ = _safe_chain(path, allow_missing_tail=True)
+    try:
+        absolute.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        raise ArtifactError("the research artifact store directory is unavailable") from exc
+    return _require_safe_directory(absolute)
+
+
+def _optional_safe_file(path):
+    absolute, info = _safe_chain(path, allow_missing_leaf=True)
+    if info is None:
+        return absolute, None
+    if not stat.S_ISREG(info.st_mode):
+        raise ArtifactError("a filesystem boundary is not a regular file")
+    return absolute, info
+
+
+def _reject_unsafe_existing(path):
+    """Reject link-like components while leaving type errors to the caller."""
+    return _safe_chain(path, allow_missing_leaf=True)
+
+
+def _read_safe_bytes(path, *, expected=None):
+    """Read one regular file through one fd and reject identity changes."""
+    absolute, before = _safe_chain(path)
+    if before is None or not stat.S_ISREG(before.st_mode):
+        raise ArtifactError("an explicit input has an unsupported file type")
+    if expected is not None and _file_snapshot(before) != expected:
+        raise ArtifactError("an explicit input changed before it could be read")
+
+    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_CLOEXEC", 0)
+    flags |= getattr(os, "O_NOFOLLOW", 0)
+    try:
+        fd = os.open(str(absolute), flags)
+    except OSError as exc:
+        raise ArtifactError("an explicit input could not be opened safely") from exc
+    try:
+        opened = os.fstat(fd)
+        if not stat.S_ISREG(opened.st_mode) or _file_identity(opened) != _file_identity(before):
+            raise ArtifactError("an explicit input changed before it could be read")
+        chunks = []
+        while True:
+            chunk = os.read(fd, 65536)
+            if not chunk:
+                break
+            chunks.append(chunk)
+        after_read = os.fstat(fd)
+        if _file_snapshot(after_read) != _file_snapshot(opened):
+            raise ArtifactError("an explicit input changed while it was being read")
+    finally:
+        os.close(fd)
+
+    _, after_path = _safe_chain(absolute)
+    if after_path is None or _file_snapshot(after_path) != _file_snapshot(opened):
+        raise ArtifactError("an explicit input changed while it was being read")
+    return b"".join(chunks)
+
+
+def _input_kind(path):
+    _, info = _safe_chain(path)
+    if stat.S_ISREG(info.st_mode):
+        return "file", info
+    if stat.S_ISDIR(info.st_mode):
+        return "directory", info
+    raise ArtifactError("an explicit input has an unsupported file type")
+
+
+def _directory_json_files(root, root_info):
+    files = []
+    pending = [(Path(root), _file_identity(root_info))]
+    while pending:
+        directory, expected_identity = pending.pop()
+        _, before = _require_safe_directory(directory)
+        if _file_identity(before) != expected_identity:
+            raise ArtifactError("an explicit input directory changed during traversal")
+        try:
+            with os.scandir(directory) as entries:
+                children = sorted(entries, key=lambda entry: entry.name.casefold())
+        except OSError as exc:
+            raise ArtifactError("an explicit input directory is unreadable") from exc
+        _, after = _require_safe_directory(directory)
+        if _file_identity(after) != expected_identity:
+            raise ArtifactError("an explicit input directory changed during traversal")
+        directories = []
+        for entry in children:
+            try:
+                info = entry.stat(follow_symlinks=False)
+            except OSError as exc:
+                raise ArtifactError("an explicit input entry is unavailable") from exc
+            path = Path(entry.path)
+            if _path_is_link_or_reparse(path, info):
+                raise ArtifactError("explicit inputs cannot contain links or reparse points")
+            _, bound_info = _safe_chain(path)
+            if stat.S_ISDIR(bound_info.st_mode):
+                directories.append((path, _file_identity(bound_info)))
+            elif stat.S_ISREG(bound_info.st_mode) and entry.name.casefold().endswith(".json"):
+                files.append((path, _file_snapshot(bound_info)))
+        pending.extend(reversed(directories))
+    return files
+
+
+def _input_files(inputs, target):
+    files = []
+    seen = set()
+    target_key = _path_key(target)
+    for supplied in inputs:
+        path = Path(supplied)
+        kind, info = _input_kind(path)
+        candidates = (
+            [(path, _file_snapshot(info))]
+            if kind == "file"
+            else _directory_json_files(path, info)
+        )
+        for candidate, snapshot in candidates:
+            key = _path_key(candidate)
+            if key == target_key or key in seen:
+                continue
+            seen.add(key)
+            files.append((candidate, snapshot))
+    if not files:
+        raise ArtifactError("no JSON artifact files found in the explicit inputs")
+    return files
+
+
+def ingest(inputs, memory_dir=None):
+    """Validate all inputs, then atomically merge them. Return (added, updated, total)."""
+    memdir = Path(memory_dir) if memory_dir is not None else M.mem_dir(C.data_root())
+    target = memdir / M.RESEARCH_ARTIFACTS
+    files = _input_files(inputs, target)
+
+    incoming = {}
+    for index, (source, snapshot) in enumerate(files, 1):
+        try:
+            document = json.loads(_read_safe_bytes(source, expected=snapshot).decode("utf-8"))
+            record = normalize_artifact(document)
+        except (OSError, UnicodeError) as exc:
+            raise ArtifactError(
+                f"invalid artifact input #{index}: input could not be read as UTF-8 JSON"
+            ) from exc
+        except json.JSONDecodeError as exc:
+            raise ArtifactError(f"invalid artifact input #{index}: invalid JSON") from exc
+        except ArtifactError as exc:
+            raise ArtifactError(f"invalid artifact input #{index}: {exc}") from exc
+        artifact_id = record["artifact_id"]
+        if artifact_id in incoming and incoming[artifact_id] != record:
+            raise ArtifactError(f"conflicting artifacts share stable ID {artifact_id}")
+        incoming[artifact_id] = record
+
+    try:
+        with _store_lock(target) as store_lock:
+            store_lock.validate()
+            existing = _load_existing(target)
+            store_lock.validate()
+            merged = {record["artifact_id"]: record for record in existing}
+            added = sum(artifact_id not in merged for artifact_id in incoming)
+            updated = sum(artifact_id in merged and merged[artifact_id] != record
+                          for artifact_id, record in incoming.items())
+            merged.update(incoming)
+            artifacts = [merged[key] for key in sorted(merged)]
+            if artifacts != existing:
+                store_lock.validate()
+                _atomic_write_json(
+                    target,
+                    {"schema_version": STORE_VERSION, "artifacts": artifacts},
+                )
+                store_lock.validate()
+            return added, updated, len(artifacts)
+    except (WorkerBusy, WorkerStateError) as exc:
+        raise ArtifactError("research artifact store is busy or its lock is unsafe") from exc
+    except OSError as exc:
+        raise ArtifactError("research artifact store could not be read or updated") from exc
+
+
+def _parser():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("inputs", nargs="+", help="Explicit JSON artifact file(s) or directory/directories")
+    parser.add_argument("--memory-dir", type=Path, help="Override the destination memory directory")
+    return parser
+
+
+def main(argv=None):
+    args = _parser().parse_args(argv)
+    try:
+        added, updated, total = ingest(args.inputs, args.memory_dir)
+    except ArtifactError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 2
+    print(f"research artifacts ingested: added={added} updated={updated} total={total}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/provenance/ingest/vault.py b/provenance/ingest/vault.py
index fe52901..7cf64ee 100644
--- a/provenance/ingest/vault.py
+++ b/provenance/ingest/vault.py
@@ -26,6 +26,7 @@
 """
 import json
 import os
+import re
 import sys
 from datetime import datetime, timezone
 from pathlib import Path
@@ -36,6 +37,47 @@
 # project notes can carry sensitive decisions in the body; keep a short,
 # sanitized excerpt for "latest status" context without storing the whole log.
 BODY_CHARS = 800
+PROGRESS_CHARS = 4000
+PROGRESS_BEGIN = "<!-- scriptorium:progress-log:begin -->"
+PROGRESS_END = "<!-- scriptorium:progress-log:end -->"
+PROGRESS_ENTRY_RE = re.compile(r"<!--\s*entry:[^>\r\n]+-->")
+
+
+def _progress_excerpt(body):
+    """Keep the newest generated progress separately from the short body excerpt."""
+    begin = body.find(PROGRESS_BEGIN)
+    if begin < 0:
+        return ""
+    end = body.find(PROGRESS_END, begin + len(PROGRESS_BEGIN))
+    if end < 0:
+        return ""
+    region = body[begin + len(PROGRESS_BEGIN):end]
+    if len(region) > PROGRESS_CHARS:
+        matches = list(PROGRESS_ENTRY_RE.finditer(region))
+        if not matches:
+            region = region[:PROGRESS_CHARS]
+        else:
+            prefix = region[:matches[0].start()]
+            blocks = []
+            used = len(prefix)
+            for index, match in enumerate(matches):
+                block_end = (
+                    matches[index + 1].start()
+                    if index + 1 < len(matches)
+                    else len(region)
+                )
+                block = region[match.start():block_end]
+                if used + len(block) > PROGRESS_CHARS:
+                    break
+                blocks.append(block)
+                used += len(block)
+            if blocks:
+                region = prefix + "".join(blocks)
+            else:
+                # A single pathological entry may exceed the bound. Keep its
+                # newest prefix instead of retaining older entries from the tail.
+                region = region[:PROGRESS_CHARS]
+    return f"{PROGRESS_BEGIN}{region}{PROGRESS_END}"
 
 
 def _collect(note_dir, glob, schema, id_field):
@@ -60,6 +102,9 @@ def _collect(note_dir, glob, schema, id_field):
         excerpt = C.clip(body, BODY_CHARS)
         if excerpt:
             rec["body"] = excerpt
+        progress_excerpt = _progress_excerpt(body)
+        if progress_excerpt:
+            rec["_progress_log"] = progress_excerpt
         rec["_source_file"] = md.name
         out[str(rid)] = sanitize_json(rec)
     return out
diff --git a/provenance/mcp_server.py b/provenance/mcp_server.py
index bf145a7..3bd00fe 100644
--- a/provenance/mcp_server.py
+++ b/provenance/mcp_server.py
@@ -15,6 +15,7 @@
   - list_topics()               category counts from the organize catalog
   - get_portfolio()             the research portfolio (总纲): projects + stage + next steps
   - get_current_context(project_id)  one project's literature + status
+  - get_context_capsule(project_id)  bounded, approval-aware resume context
 
 Register (Claude Code example, run from your data dir or set PROVENANCE_HOME):
   claude mcp add provenance -- prov-mcp
@@ -26,10 +27,10 @@
 import sys
 from collections import Counter
 
-from provenance import common as C, memory as M
+from provenance import common as C, context as CX, memory as M
 from provenance.search import core as S
 
-VERSION = "0.17.0"
+VERSION = "0.18.0"
 PROTOCOL = "2024-11-05"
 
 TOOLS = [
@@ -53,6 +54,11 @@
      "inputSchema": {"type": "object",
                      "properties": {"project_id": {"type": "string"}},
                      "required": ["project_id"]}},
+    {"name": "get_context_capsule",
+     "description": "A bounded, read-only resume capsule for one project. Human/approved state is authoritative; literature artifact leads are explicitly reference-only.",
+     "inputSchema": {"type": "object",
+                     "properties": {"project_id": {"type": "string"}},
+                     "required": ["project_id"]}},
 ]
 
 
@@ -183,8 +189,19 @@ def _get_current_context(args):
     return "\n".join(out)
 
 
+def _get_context_capsule(args):
+    pid = (args.get("project_id") or "").strip()
+    if not pid:
+        return "Provide a 'project_id' (see get_portfolio)."
+    try:
+        return CX.render_markdown(CX.build_capsule(pid))
+    except CX.ProjectNotFoundError as exc:
+        return str(exc)
+
+
 HANDLERS = {"search_brain": _search_brain, "get_profile": _get_profile, "list_topics": _list_topics,
-            "get_portfolio": _get_portfolio, "get_current_context": _get_current_context}
+            "get_portfolio": _get_portfolio, "get_current_context": _get_current_context,
+            "get_context_capsule": _get_context_capsule}
 
 
 def handle(req):
diff --git a/provenance/memory.py b/provenance/memory.py
index d49894f..437dace 100644
--- a/provenance/memory.py
+++ b/provenance/memory.py
@@ -11,9 +11,10 @@
 
     memory/library.json      reference library snapshot (literature)
     memory/projects.json     project portfolio (project/1.0 frontmatter)
+    memory/research-artifacts.json  parsed papers, reading notes, reviews, lineage
 
-Written by the ingesters (provenance.ingest.library / .vault); read by the
-read-only MCP tools (get_portfolio / get_current_context) and by search indexing.
+Written by the library, vault and research-artifact ingesters; read by the
+read-only MCP tools, Context Capsule builder and search indexing.
 Stdlib only, no network — same privacy posture as the rest of the pipeline.
 
 Also home to a minimal YAML-frontmatter parser (a stdlib subset, NOT PyYAML) so
@@ -28,6 +29,7 @@
 LIBRARY = "library.json"
 PROJECTS = "projects.json"
 APPLIED_CLAIMS = "applied_claims.json"   # sync-layer M3 approved high-value claims
+RESEARCH_ARTIFACTS = "research-artifacts.json"
 
 # high-value claim fields an approved session-summary overlays onto a project record
 APPLIED_FIELDS = ("status", "stage", "next_actions", "conclusion", "blocked_by")
@@ -51,6 +53,17 @@ def load_library(root=None):
     return items if isinstance(items, list) else []
 
 
+def load_research_artifacts(root=None):
+    """Return sanitized literature-research artifact records, or an empty list.
+
+    Each record contains ``artifact_id``, ``artifact_type``, ``schema_version``
+    and the original contract object under ``content``. The research ingester
+    guarantees stable IDs, path removal and privacy sanitization before write.
+    """
+    artifacts = _read(mem_dir(root) / RESEARCH_ARTIFACTS).get("artifacts")
+    return artifacts if isinstance(artifacts, list) else []
+
+
 def load_applied_claims(root=None):
     """Return {project_id: approved high-value claims} (sync-layer M3 overlay).
 
diff --git a/provenance/search/core.py b/provenance/search/core.py
index 0f9f74c..e4ba7f1 100644
--- a/provenance/search/core.py
+++ b/provenance/search/core.py
@@ -112,6 +112,24 @@ def build(inbox, db=None):
             " ".join(it.get("folders") or []), it.get("readStatus") or ""] if p)
         rows.append((f"lit:{key}", "literature", str(date), "literature", title, body))
         n += 1
+    # Research artifacts are already privacy-sanitized at ingest. Index their
+    # complete contract content so paper sections, reading findings, review gaps
+    # and lineage relations are discoverable through the same local FTS surface.
+    for record in M.load_research_artifacts(inbox.parent):
+        if not isinstance(record, dict) or not isinstance(record.get("content"), dict):
+            continue
+        content = record["content"]
+        artifact_type = record.get("artifact_type") or "research-artifact"
+        artifact_id = record.get("artifact_id") or ""
+        direction = content.get("direction") if isinstance(content.get("direction"), dict) else {}
+        metadata = content.get("metadata") if isinstance(content.get("metadata"), dict) else {}
+        title = (metadata.get("title") or direction.get("query") or content.get("id")
+                 or artifact_id or "(untitled research artifact)")
+        date = content.get("created") or direction.get("created") or ""
+        body = json.dumps(content, ensure_ascii=False, sort_keys=True)
+        rows.append((f"research:{artifact_id}", artifact_type, str(date),
+                     "research-artifact", str(title), body))
+        n += 1
     con.executemany("INSERT INTO conv VALUES (?,?,?,?,?,?)", rows)
     con.commit()
     con.close()
diff --git a/provenance/synclayer/pull.py b/provenance/synclayer/pull.py
index 6e531df..87500b0 100644
--- a/provenance/synclayer/pull.py
+++ b/provenance/synclayer/pull.py
@@ -447,6 +447,37 @@ def _execute(report, workspace, root, *, run, scan_codex, codex_home, project):
         "ok" if run else "planned",
         {"pending": refresh["pending"]},
     )
+
+    # The portfolio snapshot is ingested before the worker so session resolution
+    # uses a stable view. When this run writes a progress block or applies an
+    # approval, refresh that snapshot before returning so an immediate read-only
+    # context capsule observes the just-completed writeback.
+    if run and (worker_counts["applied"] or checked):
+        try:
+            refreshed_portfolio = vault_ingest.ingest(
+                workspace, root=root, run=True
+            )
+        except (OSError, UnicodeError, ValueError):
+            for stage in report["stages"]:
+                if stage.get("id") == "portfolio":
+                    stage["status"] = "partial"
+                    break
+            _mark(
+                report,
+                "partial",
+                1,
+                "stage_failed",
+                "The project writeback completed but its memory snapshot could not be refreshed.",
+            )
+            return report
+        except Exception:
+            for stage in report["stages"]:
+                if stage.get("id") == "portfolio":
+                    stage["status"] = "error"
+                    break
+            _mark(report, "error", 2, "internal_error", "Pull failed internally.")
+            return report
+        report["summary"]["projects"] = refreshed_portfolio["count"]
     return _finish(report)
 
 
diff --git a/pyproject.toml b/pyproject.toml
index c43fbe3..9438636 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -17,7 +17,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "provenance"
-version = "0.17.0"
+version = "0.18.0"
 description = "Local-first pipeline that captures, sanitizes (secret-scrub + pseudonymize), organizes and profiles your cross-AI conversation history into a private second brain."
 requires-python = ">=3.9"
 dependencies = []
@@ -38,6 +38,7 @@ prov-ingest = "provenance.ingest.core:main"
 prov-ingest-takeout = "provenance.ingest.takeout:main"
 prov-ingest-agents = "provenance.ingest.agents:main"
 prov-ingest-library = "provenance.ingest.library:main"
+prov-ingest-research = "provenance.ingest.research:main"
 prov-ingest-vault = "provenance.ingest.vault:main"
 prov-ingest-notes = "provenance.ingest.notes:main"
 prov-ingest-openclaw = "provenance.ingest.openclaw:main"
@@ -59,6 +60,7 @@ prov-sync-approve = "provenance.synclayer.approvals:main"
 prov-sync-scan-agents = "provenance.synclayer.scan_agents:main"
 prov-sync-pull = "provenance.synclayer.pull:main"
 prov-status = "provenance.status:main"
+prov-context = "provenance.context:main"
 
 [tool.setuptools]
 packages = ["provenance", "provenance.ingest", "provenance.organize", "provenance.export", "provenance.search", "provenance.synclayer"]
diff --git a/tests/test_context_capsule.py b/tests/test_context_capsule.py
new file mode 100644
index 0000000..482d5e2
--- /dev/null
+++ b/tests/test_context_capsule.py
@@ -0,0 +1,489 @@
+#!/usr/bin/env python3
+"""Context Capsule contract tests; synthetic files only, no private data."""
+import contextlib
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+from provenance import context  # noqa: E402
+from provenance import mcp_server  # noqa: E402
+
+
+PROJECT_ID = "synthetic-alpha"
+CITEKEY = "example2100SyntheticAlpha"
+
+
+class TestContextCapsule(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        memory = self.root / "memory"
+        memory.mkdir()
+
+        project = {
+            "schema_version": "project/1.0",
+            "project_id": PROJECT_ID,
+            "title": "[SYNTHETIC] Alpha project",
+            "status": "planned",
+            "stage": "intake",
+            "priority": "medium",
+            "next_actions": ["Human-authored fallback action"],
+            "blocked_by": "",
+            "linked_literature": ["DEMO0001", "Synthetic/Alpha"],
+            "linked_repo": r"C:\Private\SyntheticAlpha",
+            "updated": "2100-01-01",
+            "body": (
+                "Research goal: Verify a generated alpha workflow using "
+                r"C:\Private\SyntheticAlpha\inputs.json and /private/synthetic/input.json"
+                "\n\nHuman-authored project note.\n"
+                "<!-- scriptorium:progress-log:begin -->\n"
+                "<!-- entry:session-new#approved -->\n"
+                "**2100-01-03** · approved\n"
+                "- conclusion: duplicated approved log text\n"
+                "<!-- entry:session-new -->\n"
+                "**2100-01-03** · session writeback\n"
+                "- Ran the generated benchmark\n"
+                "- Recorded the low-risk fixture count\n"
+                "<!-- scriptorium:progress-log:end -->\n"
+            ),
+        }
+        (memory / "projects.json").write_text(
+            json.dumps({"projects": {PROJECT_ID: project}}, ensure_ascii=False),
+            encoding="utf-8",
+        )
+        (memory / "applied_claims.json").write_text(
+            json.dumps({
+                "claims": {
+                    PROJECT_ID: {
+                        "status": "active",
+                        "stage": "approved validation",
+                        "next_actions": ["Run the approved synthetic check"],
+                        "conclusion": "Approved synthetic conclusion",
+                        "blocked_by": "Awaiting generated fixture",
+                    }
+                }
+            }),
+            encoding="utf-8",
+        )
+        (memory / "library.json").write_text(
+            json.dumps({
+                "items": [{
+                    "key": "DEMO0001",
+                    "citekey": CITEKEY,
+                    "title": "[SYNTHETIC] Alpha reference",
+                    "year": "2100",
+                    "readStatus": "In Progress",
+                    "folders": ["Synthetic/Alpha/Baselines"],
+                    "tldr": r"Reference fixture at C:\Private Lab\papers\alpha.pdf after path",
+                }]
+            }),
+            encoding="utf-8",
+        )
+        (memory / "research-artifacts.json").write_text(
+            json.dumps({
+                "schema_version": "research-artifacts/1.0",
+                "artifacts": [
+                    {
+                        "artifact_id": "parsed:" + CITEKEY,
+                        "artifact_type": "parsed-paper",
+                        "schema_version": "parsed-paper/1.0",
+                        "content": {
+                            "schema_version": "parsed-paper/1.0",
+                            "id": CITEKEY,
+                            "created": "2100-01-02",
+                            "parser": "synthetic",
+                            "source_pdf": r"C:\Private\papers\alpha.pdf",
+                            "metadata": {"title": "[SYNTHETIC] Parsed alpha paper"},
+                            "sections": [{"heading": "Generated"}],
+                            "references": [],
+                            "figures": [],
+                        },
+                    },
+                    {
+                        "artifact_id": "reading:" + CITEKEY,
+                        "artifact_type": "reading-note",
+                        "schema_version": "reading-note/1.0",
+                        "content": {
+                            "schema_version": "reading-note/1.0",
+                            "id": CITEKEY,
+                            "created": "2100-01-02",
+                            "read_status": "In Progress",
+                            "stages": {"glance": {"tldr": "Candidate reading interpretation"}},
+                            "sources": {"parsed_file": r"C:\Private\parsed\alpha.json"},
+                        },
+                    },
+                    {
+                        "artifact_id": "review:synthetic-alpha",
+                        "artifact_type": "review",
+                        "schema_version": "review/1.0",
+                        "content": {
+                            "schema_version": "review/1.0",
+                            "direction": {"query": "Synthetic alpha direction", "created": "2100-01-03"},
+                            "sections": [{"heading": "Overview", "prose": "Candidate prose"}],
+                            "comparison_table": {
+                                "rows": [{"citekey": CITEKEY, "cells": ["candidate"]}]
+                            },
+                            "gaps": ["Candidate gap from the literature review"],
+                            "priority_reads": [CITEKEY],
+                            "lineage_ref": r"C:\Private\lineage.json",
+                        },
+                    },
+                    {
+                        "artifact_id": "lineage:synthetic-alpha",
+                        "artifact_type": "lineage-graph",
+                        "schema_version": "lineage-graph/1.0",
+                        "content": {
+                            "schema_version": "lineage-graph/1.0",
+                            "direction": {"query": "Synthetic alpha direction", "created": "2100-01-03"},
+                            "nodes": [{"citekey": CITEKEY, "title": "Alpha"}],
+                            "edges": [],
+                        },
+                    },
+                    {
+                        "artifact_id": "lineage:related-only-by-seventh-node",
+                        "artifact_type": "lineage-graph",
+                        "schema_version": "lineage-graph/1.0",
+                        "content": {
+                            "schema_version": "lineage-graph/1.0",
+                            "direction": {"query": "Seventh-node direction", "created": "2100-01-03"},
+                            "nodes": [
+                                {"citekey": f"unrelated-{index}"} for index in range(6)
+                            ] + [{"citekey": CITEKEY}],
+                            "edges": [],
+                        },
+                    },
+                    {
+                        "artifact_id": "parsed:unrelated",
+                        "artifact_type": "parsed-paper",
+                        "schema_version": "parsed-paper/1.0",
+                        "content": {
+                            "schema_version": "parsed-paper/1.0",
+                            "id": "unrelated2100Paper",
+                            "created": "2100-01-02",
+                            "metadata": {"title": "UNRELATED ARTIFACT"},
+                        },
+                    },
+                ],
+            }, ensure_ascii=False),
+            encoding="utf-8",
+        )
+
+        drafts = self.root / "sync-state" / "drafts"
+        drafts.mkdir(parents=True)
+        (drafts / "draft.json").write_text(json.dumps({
+            "approval_state": "draft",
+            "project": PROJECT_ID,
+            "conclusion": "DRAFT MUST NOT APPEAR",
+            "next_actions": ["DRAFT ACTION MUST NOT APPEAR"],
+        }), encoding="utf-8")
+        (drafts / "rejected.json").write_text(json.dumps({
+            "approval_state": "rejected",
+            "project": PROJECT_ID,
+            "conclusion": "REJECTED MUST NOT APPEAR",
+        }), encoding="utf-8")
+        inbox = self.root / "inbox" / "codex"
+        inbox.mkdir(parents=True)
+        (inbox / "raw.json").write_text(json.dumps({
+            "chat_messages": [{"text": "RAW CHAT MUST NOT APPEAR"}],
+        }), encoding="utf-8")
+
+    def _snapshot(self):
+        return {
+            str(path.relative_to(self.root)): (path.read_bytes(), path.stat().st_mtime_ns)
+            for path in sorted(self.root.rglob("*")) if path.is_file()
+        }
+
+    def _mcp_call(self, name, arguments=None):
+        old = os.environ.get("PROVENANCE_HOME")
+        os.environ["PROVENANCE_HOME"] = str(self.root)
+        try:
+            response = mcp_server.handle({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "tools/call",
+                "params": {"name": name, "arguments": arguments or {}},
+            })
+            return response["result"]["content"][0]["text"]
+        finally:
+            if old is None:
+                os.environ.pop("PROVENANCE_HOME", None)
+            else:
+                os.environ["PROVENANCE_HOME"] = old
+
+    def test_case_insensitive_id_and_approved_fields(self):
+        capsule = context.build_capsule("SYNTHETIC-ALPHA", root=self.root)
+        self.assertEqual(capsule["project"]["project_id"], PROJECT_ID)
+        self.assertEqual(capsule["project"]["status"], "active")
+        self.assertEqual(capsule["project"]["stage"], "approved validation")
+        self.assertEqual(capsule["project"]["conclusion"], "Approved synthetic conclusion")
+        self.assertEqual(capsule["next_actions"], ["Run the approved synthetic check"])
+        self.assertEqual(capsule["trust"]["project_state"], "human_or_approved")
+        self.assertEqual(
+            capsule["trust"]["recent_progress"],
+            "auto_applied_low_risk_not_approved_claims",
+        )
+
+    def test_canonical_init_note_uses_research_intuition_not_title(self):
+        projects_path = self.root / "memory" / "projects.json"
+        document = json.loads(projects_path.read_text(encoding="utf-8"))
+        project = document["projects"][PROJECT_ID]
+        project["body"] = (
+            "# [SYNTHETIC] Alpha project\n\n"
+            "## Research intuition\n\n"
+            "Test a synthetic catalyst hypothesis.\n\n"
+            "## Research question\n\n"
+            "<!-- Define the research question after review. -->\n\n"
+            "## Progress log\n\n"
+            "<!-- scriptorium:progress-log:begin -->\n"
+            "<!-- scriptorium:progress-log:end -->\n"
+        )
+        projects_path.write_text(json.dumps(document), encoding="utf-8")
+
+        capsule = context.build_capsule(PROJECT_ID, root=self.root)
+
+        self.assertEqual(
+            capsule["project"]["goal"],
+            "Test a synthetic catalyst hypothesis.",
+        )
+
+    def test_unknown_project(self):
+        with self.assertRaises(context.ProjectNotFoundError) as raised:
+            context.build_capsule("missing", root=self.root)
+        self.assertEqual(str(raised.exception), "Project not found.")
+
+    def test_unknown_project_error_is_fixed_and_bounded(self):
+        requested = "private-" + ("x" * 50000)
+        projects = {
+            f"private-project-{index}-" + ("y" * 500): {}
+            for index in range(100)
+        }
+
+        with self.assertRaises(context.ProjectNotFoundError) as raised:
+            context._find_project(requested, projects)
+
+        message = str(raised.exception)
+        self.assertEqual(message, "Project not found.")
+        self.assertNotIn("private-", message)
+        self.assertNotIn("Known:", message)
+
+    def test_drafts_and_rejected_claims_are_not_read(self):
+        text = json.dumps(context.build_capsule(PROJECT_ID, root=self.root), ensure_ascii=False)
+        self.assertNotIn("DRAFT MUST NOT APPEAR", text)
+        self.assertNotIn("DRAFT ACTION MUST NOT APPEAR", text)
+        self.assertNotIn("REJECTED MUST NOT APPEAR", text)
+        self.assertNotIn("RAW CHAT MUST NOT APPEAR", text)
+        self.assertIn("Ran the generated benchmark", text)  # low-risk timeline is allowed
+        self.assertNotIn("duplicated approved log text", text)  # approved log block is not re-read
+
+    def test_paths_are_not_exposed(self):
+        capsule = context.build_capsule(PROJECT_ID, root=self.root)
+        text = json.dumps(capsule, ensure_ascii=False)
+        self.assertNotIn(r"C:\Private", text)
+        self.assertNotIn(r"Lab\papers\alpha.pdf", text)
+        self.assertNotIn("/private/synthetic", text)
+        self.assertNotIn("source_pdf", text)
+        self.assertNotIn("parsed_file", text)
+        self.assertNotIn("linked_repo", text)
+        self.assertIn("[local path omitted]", text)
+
+    def test_file_uris_are_omitted_without_redacting_public_urls(self):
+        public_url = "https://example.org/public-paper"
+        drive_like_public_url = "https://example.org/C:/docs/readme"
+        projects_path = self.root / "memory" / "projects.json"
+        projects = json.loads(projects_path.read_text(encoding="utf-8"))
+        project = projects["projects"][PROJECT_ID]
+        project["goal"] = (
+            "Review file:///srv/private/hidden.pdf and "
+            + public_url
+            + " with "
+            + drive_like_public_url
+        )
+        project["body"] = (
+            "<!-- scriptorium:progress-log:begin -->\n"
+            "<!-- entry:session-file-uri -->\n"
+            "**2100-01-04** - session writeback\n"
+            "- Compared file://lab-host/share/Private%20Research/result.pdf "
+            "with " + public_url + "\n"
+            "<!-- scriptorium:progress-log:end -->\n"
+        )
+        projects_path.write_text(
+            json.dumps(projects, ensure_ascii=False), encoding="utf-8"
+        )
+        claims_path = self.root / "memory" / "applied_claims.json"
+        claims = json.loads(claims_path.read_text(encoding="utf-8"))
+        claims["claims"][PROJECT_ID]["next_actions"] = [
+            "Open FILE:///C:/Users/Demo/Private%20Research/plan.md",
+            "Open file:///C:/Users/Demo/Private Research/plan.md and continue",
+            "Consult " + public_url,
+        ]
+        claims_path.write_text(
+            json.dumps(claims, ensure_ascii=False), encoding="utf-8"
+        )
+
+        capsule = context.build_capsule(PROJECT_ID, root=self.root)
+        text = json.dumps(capsule, ensure_ascii=False)
+
+        self.assertNotIn("file:", text.casefold())
+        self.assertNotIn("hidden.pdf", text)
+        self.assertNotIn("lab-host", text)
+        self.assertNotIn("Private%20Research", text)
+        self.assertNotIn("Private Research", text)
+        self.assertNotIn(r"\plan.md", text)
+        self.assertGreaterEqual(text.count("[local path omitted]"), 4)
+        self.assertIn(public_url, capsule["project"]["goal"])
+        self.assertIn(drive_like_public_url, capsule["project"]["goal"])
+        self.assertTrue(
+            any(public_url in item for item in capsule["next_actions"])
+        )
+        self.assertIn(public_url, capsule["recent_progress"][0]["items"][0])
+
+    def test_windows_user_path_is_removed_before_pii_redaction(self):
+        value = r"Inspect C:\Users\Demo\Private Research\result.json and continue"
+
+        bounded = context._bounded(value)
+
+        self.assertEqual(bounded, "Inspect [local path omitted]")
+
+    def test_parenthesized_file_uri_does_not_leak_its_tail(self):
+        value = "See (file:///C:/Users/Demo/Private Research/plan.md) and continue"
+
+        bounded = context._bounded(value)
+
+        self.assertEqual(bounded, "See ([local path omitted]) and continue")
+        self.assertNotIn("Private Research", bounded)
+        self.assertNotIn("plan.md", bounded)
+
+    def test_file_uri_path_words_are_not_treated_as_prose_boundaries(self):
+        value = (
+            "Inspect file:///C:/Users/Demo/Research and Development/"
+            "Data with Spaces/plan.md"
+        )
+
+        bounded = context._bounded(value)
+
+        self.assertEqual(bounded, "Inspect [local path omitted]")
+        self.assertNotIn("Development", bounded)
+        self.assertNotIn("plan.md", bounded)
+
+    def test_posix_path_with_spaces_does_not_leak_its_tail(self):
+        value = "Inspect /home/demo/Private Research/plan.md and continue"
+
+        bounded = context._bounded(value)
+
+        self.assertEqual(bounded, "Inspect [local path omitted]")
+        self.assertNotIn("Research", bounded)
+        self.assertNotIn("plan.md", bounded)
+
+    def test_multiline_local_paths_do_not_leak_their_first_line(self):
+        value = (
+            "See file:///srv/private/hidden paper.pdf\n"
+            "Then inspect /home/demo/Private Research/plan.md\n"
+            "| file:///srv/private/table plan.md | note |\n"
+            "Next line"
+        )
+
+        bounded = context._bounded(value)
+
+        self.assertNotIn("file:", bounded.casefold())
+        self.assertNotIn("hidden paper.pdf", bounded)
+        self.assertNotIn("Private Research", bounded)
+        self.assertNotIn("plan.md", bounded)
+        self.assertNotIn("table plan.md", bounded)
+        self.assertNotIn("Demo", bounded)
+        self.assertNotIn("Private Research", bounded)
+        self.assertNotIn("result.json", bounded)
+
+    def test_related_artifacts_are_reference_only(self):
+        capsule = context.build_capsule(PROJECT_ID, root=self.root)
+        kinds = {item["kind"] for item in capsule["research_artifacts"]}
+        self.assertEqual(kinds, {"parsed-paper", "reading-note", "review", "lineage-graph"})
+        self.assertTrue(any(
+            item["id"] == "lineage:related-only-by-seventh-node"
+            for item in capsule["research_artifacts"]
+        ))
+        self.assertTrue(all(item["trust"] == "reference_only"
+                            for item in capsule["research_artifacts"]))
+        self.assertNotIn("UNRELATED ARTIFACT", json.dumps(capsule, ensure_ascii=False))
+        self.assertEqual(capsule["reference_leads"]["gaps"],
+                         ["Candidate gap from the literature review"])
+        rendered = context.render_markdown(capsule)
+        self.assertIn("not approved claims", rendered)
+
+    def test_idempotent_bounded_and_read_only(self):
+        before = self._snapshot()
+        first = context.build_capsule(PROJECT_ID, root=self.root)
+        first_md = context.render_markdown(first)
+        second = context.build_capsule(PROJECT_ID, root=self.root)
+        after = self._snapshot()
+        self.assertEqual(first, second)
+        self.assertEqual(first_md, context.render_markdown(second))
+        self.assertLessEqual(len(first_md), context.MAX_MARKDOWN_CHARS)
+        self.assertLessEqual(
+            len(json.dumps(first, ensure_ascii=False, indent=2).encode("utf-8")),
+            context.MAX_JSON_BYTES,
+        )
+        oversized = json.loads(json.dumps(first))
+        oversized["next_actions"] = ["x" * 5000] * 10
+        rendered = context.render_markdown(oversized)
+        self.assertLessEqual(len(rendered), context.MAX_MARKDOWN_CHARS)
+        self.assertIn("Capsule truncated", rendered)
+        self.assertEqual(before, after)
+
+    def test_mcp_and_json_cli(self):
+        names = {tool["name"] for tool in mcp_server.TOOLS}
+        self.assertIn("get_context_capsule", names)
+        text = self._mcp_call("get_context_capsule", {"project_id": PROJECT_ID})
+        self.assertIn("Context Capsule", text)
+        self.assertIn("reference only", text)
+
+        stdout = io.StringIO()
+        old = os.environ.get("PROVENANCE_HOME")
+        os.environ["PROVENANCE_HOME"] = str(self.root)
+        try:
+            with contextlib.redirect_stdout(stdout):
+                self.assertEqual(context.main(["--project", PROJECT_ID, "--json"]), 0)
+        finally:
+            if old is None:
+                os.environ.pop("PROVENANCE_HOME", None)
+            else:
+                os.environ["PROVENANCE_HOME"] = old
+        payload = json.loads(stdout.getvalue())
+        self.assertEqual(payload["capsule_version"], context.CAPSULE_VERSION)
+        self.assertEqual(payload["project"]["project_id"], PROJECT_ID)
+        self.assertLessEqual(
+            len(stdout.getvalue().rstrip("\n").encode("utf-8")),
+            context.MAX_JSON_BYTES,
+        )
+
+    def test_json_budget_counts_pretty_printed_utf8_bytes(self):
+        capsule = context.build_capsule(PROJECT_ID, root=self.root)
+        template = capsule["research_artifacts"][0]
+        capsule["research_artifacts"] = [
+            {
+                **template,
+                "id": f"artifact-{index}",
+                "summary": "界" * 300,
+            }
+            for index in range(20)
+        ]
+
+        context._fit_json(capsule)
+        rendered = json.dumps(capsule, ensure_ascii=False, indent=2)
+
+        self.assertLessEqual(
+            len(rendered.encode("utf-8")),
+            context.MAX_JSON_BYTES,
+        )
+        self.assertTrue(capsule["limits"]["truncated"])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)
diff --git a/tests/test_provenance.py b/tests/test_provenance.py
index 7e627cb..f0ea64d 100644
--- a/tests/test_provenance.py
+++ b/tests/test_provenance.py
@@ -280,7 +280,8 @@ def test_initialize_tools_list_and_notification(self):
         tl = M.handle({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
         names = {t["name"] for t in tl["result"]["tools"]}
         self.assertEqual(names, {"search_brain", "get_profile", "list_topics",
-                                 "get_portfolio", "get_current_context"})
+                                 "get_portfolio", "get_current_context",
+                                 "get_context_capsule"})
         self.assertIsNone(M.handle({"jsonrpc": "2.0", "method": "notifications/initialized"}))
 
     def test_tools_call_get_profile(self):
@@ -600,6 +601,59 @@ def test_ingest_projects(self):
         self.assertEqual(projects["synthetic-xq17-calibration"]["next_actions"],
                          ["Review synthetic benchmark results", "Complete synthetic ablation"])
 
+    def test_keeps_recent_progress_beyond_the_short_body_excerpt(self):
+        from provenance.ingest import vault as IV
+        td = tempfile.TemporaryDirectory(); self.addCleanup(td.cleanup)
+        root = Path(td.name)
+        vault = self._vault(root)
+        note = vault / "Projects" / "ml.md"
+        note.write_text(
+            PROJECT_MD
+            + ("Long human context. " * 100)
+            + "\n<!-- scriptorium:progress-log:begin -->\n"
+            + "<!-- entry:synthetic-session -->\n"
+            + "**2100-01-03** - session writeback\n"
+            + "- newest synthetic progress\n"
+            + "<!-- scriptorium:progress-log:end -->\n",
+            encoding="utf-8",
+        )
+
+        IV.ingest(vault, root=root, run=True)
+        project = C_mem.load_projects(root)["synthetic-xq17-calibration"]
+
+        self.assertNotIn("newest synthetic progress", project["body"])
+        self.assertIn("newest synthetic progress", project["_progress_log"])
+
+    def test_progress_excerpt_keeps_newest_complete_entries(self):
+        from provenance.ingest import vault as IV
+
+        newest = (
+            "\n<!-- entry:newest -->\n"
+            "**2100-01-03** - session writeback\n"
+            "- newest synthetic progress\n"
+        )
+        old_blocks = "".join(
+            "<!-- entry:old-{index} -->\n"
+            "**2099-01-01** - session writeback\n"
+            "- old synthetic progress {index} {padding}\n".format(
+                index=index,
+                padding="x" * 500,
+            )
+            for index in range(10)
+        )
+        excerpt = IV._progress_excerpt(
+            IV.PROGRESS_BEGIN + newest + old_blocks + IV.PROGRESS_END
+        )
+
+        self.assertLessEqual(
+            len(excerpt) - len(IV.PROGRESS_BEGIN) - len(IV.PROGRESS_END),
+            IV.PROGRESS_CHARS,
+        )
+        self.assertIn("<!-- entry:newest -->", excerpt)
+        self.assertIn("newest synthetic progress", excerpt)
+        self.assertNotIn("<!-- entry:old-9 -->", excerpt)
+        self.assertTrue(excerpt.rstrip().endswith(IV.PROGRESS_END))
+
 
 class TestSearchLiterature(unittest.TestCase):
     def test_library_indexed_alongside_conversations(self):
diff --git a/tests/test_research_artifacts.py b/tests/test_research_artifacts.py
new file mode 100644
index 0000000..610bae2
--- /dev/null
+++ b/tests/test_research_artifacts.py
@@ -0,0 +1,523 @@
+#!/usr/bin/env python3
+"""Tests for explicit, local-only research-artifact ingestion."""
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+from provenance import memory as M  # noqa: E402
+from provenance.ingest import research as R  # noqa: E402
+from provenance.search import core as S  # noqa: E402
+
+
+def _remove_directory_link(path):
+    try:
+        if os.name == "nt":
+            os.rmdir(path)
+        else:
+            path.unlink()
+    except FileNotFoundError:
+        pass
+
+
+def _make_directory_link(testcase, link, target):
+    if os.name == "nt":
+        result = subprocess.run(
+            ["cmd", "/d", "/c", "mklink", "/J", str(link), str(target)],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+        if result.returncode != 0:
+            testcase.skipTest("directory junctions are unavailable")
+    else:
+        try:
+            link.symlink_to(target, target_is_directory=True)
+        except OSError as exc:
+            testcase.skipTest(f"directory symlinks are unavailable: {exc}")
+    testcase.addCleanup(_remove_directory_link, link)
+
+
+def _fixtures():
+    secret = "-".join(("sk", "ant", "api03", "ABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890"))
+    return [
+        {
+            "schema_version": "parsed-paper/1.0",
+            "id": "demo2026Calibration",
+            "created": "2026-07-22T10:00:00Z",
+            "parser": "synthetic",
+            "source_pdf": r"C:\Users\demo\Private Research\papers\demo.pdf",
+            "metadata": {"title": "Synthetic microthermometry calibration",
+                         "authors": ["Example Author"],
+                         "abstract": "Private cache: /srv/research/private/demo.txt",
+                         "local_uri": "file:///srv/private/hidden-paper.pdf",
+                         r"C:\Private Lab\metadata.json": "path-shaped object key",
+                         secret: "credential-shaped object key"},
+            "sections": [{"heading": "Method", "level": 1,
+                          "text": f"Run the synthetic assay; token {secret}; "
+                                  r"log C:\Private Lab\runs\demo.log after execution"}],
+            "references": [],
+        },
+        {
+            "schema_version": "reading-note/1.0",
+            "id": "demo2026Calibration",
+            "created": "2026-07-22T10:10:00Z",
+            "read_status": "In Progress",
+            "stages": {"glance": {"tldr": "A synthetic geothermometry orientation",
+                                    "tags": ["petrology"],
+                                    "key_findings": ["Calibration is deterministic"]}},
+            "sources": {"parsed_file": r"D:\private\parsed\demo.json"},
+        },
+        {
+            "schema_version": "review/1.0",
+            "direction": {"query": "Synthetic thermobarometry", "created": "2026-07-22T10:20:00Z"},
+            "sections": [{"heading": "Comparison", "prose": "Bibliometric synthesis of synthetic evidence."}],
+            "gaps": ["Independent validation is missing"],
+            "priority_reads": ["demo2026Calibration"],
+            "lineage_ref": r"C:\Users\demo\Private Research\lineage.json",
+        },
+        {
+            "schema_version": "lineage-graph/1.0",
+            "direction": {"query": "Synthetic thermobarometry",
+                          "scope_method": "synthetic own-library citation match",
+                          "created": "2026-07-22T10:30:00Z"},
+            "nodes": [{"citekey": "demo2026Calibration", "title": "Synthetic lineage anchor"}],
+            "edges": [{"from": "demo2026Calibration", "to": "baseline2018",
+                       "relation": "cites", "evidence": "Synthetic reference"}],
+        },
+    ]
+
+
+class TestResearchArtifactIngest(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.inputs = self.root / "explicit-inputs"
+        self.inputs.mkdir()
+        for index, document in enumerate(_fixtures()):
+            (self.inputs / f"artifact-{index}.json").write_text(
+                json.dumps(document, ensure_ascii=False), encoding="utf-8")
+        self.memdir = self.root / "memory"
+
+    def test_ingests_all_four_contracts_and_loads_them(self):
+        self.assertEqual(R.ingest([self.inputs], self.memdir), (4, 0, 4))
+        artifacts = M.load_research_artifacts(self.root)
+        self.assertEqual(len(artifacts), 4)
+        self.assertEqual({item["artifact_type"] for item in artifacts},
+                         {"parsed-paper", "reading-note", "review", "lineage-graph"})
+        self.assertEqual(len({item["artifact_id"] for item in artifacts}), 4)
+        self.assertTrue(all(item["content"]["schema_version"] == item["schema_version"]
+                            for item in artifacts))
+
+    def test_repeated_ingest_is_byte_stable_and_has_no_duplicates(self):
+        self.assertEqual(R.ingest([self.inputs], self.memdir), (4, 0, 4))
+        target = self.memdir / M.RESEARCH_ARTIFACTS
+        first = target.read_bytes()
+        self.assertEqual(R.ingest([self.inputs], self.memdir), (0, 0, 4))
+        self.assertEqual(target.read_bytes(), first)
+        self.assertEqual(len(M.load_research_artifacts(self.root)), 4)
+
+    def test_case_variant_citekeys_remain_distinct_across_ingest_runs(self):
+        upper = self.root / "upper.json"
+        lower = self.root / "lower.json"
+        first = {
+            "schema_version": "parsed-paper/1.0",
+            "id": "Example2100CaseKey",
+            "created": "2100-01-01T00:00:00Z",
+        }
+        second = dict(first, id="example2100casekey")
+        upper.write_text(json.dumps(first), encoding="utf-8")
+        lower.write_text(json.dumps(second), encoding="utf-8")
+
+        self.assertEqual(R.ingest([upper], self.memdir), (1, 0, 1))
+        self.assertEqual(R.ingest([lower], self.memdir), (1, 0, 2))
+        artifacts = M.load_research_artifacts(self.root)
+        self.assertEqual(
+            {record["content"]["id"] for record in artifacts},
+            {"Example2100CaseKey", "example2100casekey"},
+        )
+        self.assertEqual(len({record["artifact_id"] for record in artifacts}), 2)
+
+    def test_case_variant_lineage_citekeys_remain_distinct_without_query(self):
+        upper = _fixtures()[3]
+        upper["direction"]["query"] = ""
+        upper["nodes"][0]["citekey"] = "Example2100CaseKey"
+        lower = json.loads(json.dumps(upper))
+        lower["nodes"][0]["citekey"] = "example2100casekey"
+        upper_path = self.root / "lineage-upper.json"
+        lower_path = self.root / "lineage-lower.json"
+        upper_path.write_text(json.dumps(upper), encoding="utf-8")
+        lower_path.write_text(json.dumps(lower), encoding="utf-8")
+
+        self.assertEqual(R.ingest([upper_path, lower_path], self.memdir), (2, 0, 2))
+        artifacts = M.load_research_artifacts(self.root)
+        self.assertEqual(len({record["artifact_id"] for record in artifacts}), 2)
+
+    def test_review_and_lineage_identity_include_created(self):
+        review_a = _fixtures()[2]
+        review_b = json.loads(json.dumps(review_a))
+        review_b["direction"]["created"] = "2100-02-01T00:00:00Z"
+        review_b["sections"][0]["prose"] = "A later synthetic synthesis."
+        lineage_a = _fixtures()[3]
+        lineage_b = json.loads(json.dumps(lineage_a))
+        lineage_b["direction"]["created"] = "2100-02-01T00:00:00Z"
+        paths = []
+        for index, document in enumerate((review_a, review_b, lineage_a, lineage_b)):
+            path = self.root / f"versioned-{index}.json"
+            path.write_text(json.dumps(document), encoding="utf-8")
+            paths.append(path)
+
+        self.assertEqual(R.ingest(paths, self.memdir), (4, 0, 4))
+        artifacts = M.load_research_artifacts(self.root)
+        self.assertEqual(len({record["artifact_id"] for record in artifacts}), 4)
+
+        review_b["sections"][0]["prose"] = "Updated content at the same timestamp."
+        paths[1].write_text(json.dumps(review_b), encoding="utf-8")
+        self.assertEqual(R.ingest([paths[1]], self.memdir), (0, 1, 4))
+
+    def test_invalid_schema_never_replaces_existing_store(self):
+        R.ingest([self.inputs], self.memdir)
+        target = self.memdir / M.RESEARCH_ARTIFACTS
+        before = target.read_bytes()
+        bad = self.root / "bad.json"
+        bad.write_text(json.dumps({"schema_version": "parsed-paper/2.0",
+                                   "id": "bad", "created": "2026-07-22"}), encoding="utf-8")
+        with self.assertRaises(R.ArtifactError):
+            R.ingest([bad], self.memdir)
+        self.assertEqual(target.read_bytes(), before)
+
+    def test_persisted_store_has_no_source_path_absolute_path_or_credential(self):
+        R.ingest([self.inputs], self.memdir)
+        text = (self.memdir / M.RESEARCH_ARTIFACTS).read_text(encoding="utf-8")
+        self.assertNotIn(str(self.inputs), text)
+        self.assertNotIn("Private Research", text)
+        self.assertNotIn("C:\\", text)
+        self.assertNotIn("D:\\", text)
+        self.assertNotIn("/srv/research", text)
+        self.assertNotIn("Private Lab", text)
+        self.assertNotIn("sk-" + "ant-api03", text)
+        self.assertIn("[REDACTED", text)
+
+    def test_concurrent_store_lock_fails_closed(self):
+        target = self.memdir / M.RESEARCH_ARTIFACTS
+        with R._store_lock(target):
+            with self.assertRaisesRegex(R.ArtifactError, "store is busy"):
+                R.ingest([self.inputs], self.memdir)
+        self.assertFalse(target.exists())
+
+    def test_independent_process_writers_do_not_lose_updates(self):
+        left = self.root / "left"
+        right = self.root / "right"
+        left.mkdir()
+        right.mkdir()
+        (left / "artifact.json").write_text(
+            json.dumps(_fixtures()[0], ensure_ascii=False), encoding="utf-8"
+        )
+        (right / "artifact.json").write_text(
+            json.dumps(_fixtures()[2], ensure_ascii=False), encoding="utf-8"
+        )
+        worker = """
+import sys
+import time
+from pathlib import Path
+from provenance.ingest import research
+
+source = Path(sys.argv[1])
+memory = Path(sys.argv[2])
+ready = Path(sys.argv[3])
+start = Path(sys.argv[4])
+original_load_existing = research._load_existing
+
+def slow_load_existing(path):
+    time.sleep(0.2)
+    return original_load_existing(path)
+
+research._load_existing = slow_load_existing
+ready.write_text("ready", encoding="utf-8")
+while not start.exists():
+    time.sleep(0.005)
+busy = 0
+for attempt in range(100):
+    try:
+        research.ingest([source], memory)
+        print(busy)
+        raise SystemExit(0)
+    except research.ArtifactError as exc:
+        if "store is busy" not in str(exc) or attempt == 99:
+            raise
+        busy += 1
+        time.sleep(0.01)
+"""
+        environment = dict(os.environ)
+        environment["PYTHONPATH"] = str(ROOT)
+        ready_paths = [self.root / "ready-left", self.root / "ready-right"]
+        start = self.root / "start"
+        processes = [
+            subprocess.Popen(
+                [
+                    sys.executable,
+                    "-c",
+                    worker,
+                    str(source),
+                    str(self.memdir),
+                    str(ready),
+                    str(start),
+                ],
+                cwd=ROOT,
+                env=environment,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            )
+            for source, ready in zip((left, right), ready_paths)
+        ]
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline and not all(path.exists() for path in ready_paths):
+            if any(process.poll() is not None for process in processes):
+                break
+            time.sleep(0.01)
+        both_ready = all(path.exists() for path in ready_paths)
+        start.write_text("start", encoding="utf-8")
+        failures = []
+        busy_counts = []
+        for process in processes:
+            stdout, stderr = process.communicate(timeout=30)
+            if process.returncode != 0:
+                failures.append((process.returncode, stdout, stderr))
+            elif stdout.strip():
+                busy_counts.append(int(stdout.strip().splitlines()[-1]))
+        self.assertTrue(both_ready, failures)
+        self.assertEqual(failures, [])
+        self.assertGreaterEqual(sum(busy_counts), 1)
+        artifacts = M.load_research_artifacts(self.root)
+        self.assertEqual(
+            {item["artifact_type"] for item in artifacts},
+            {"parsed-paper", "review"},
+        )
+        self.assertEqual(len(artifacts), 2)
+
+    def test_invalid_nested_shape_never_replaces_existing_store(self):
+        R.ingest([self.inputs], self.memdir)
+        target = self.memdir / M.RESEARCH_ARTIFACTS
+        before = target.read_bytes()
+        bad = self.root / "bad-review.json"
+        bad.write_text(json.dumps({
+            "schema_version": "review/1.0",
+            "direction": {"query": "bad", "created": "2026-07-22"},
+            "sections": [{"heading": "Missing prose"}],
+        }), encoding="utf-8")
+        with self.assertRaises(R.ArtifactError):
+            R.ingest([bad], self.memdir)
+        self.assertEqual(target.read_bytes(), before)
+
+    def test_explicit_file_symlink_is_rejected_without_reading_target(self):
+        outside = self.root / "outside.json"
+        outside.write_text(json.dumps(_fixtures()[0]), encoding="utf-8")
+        linked = self.inputs / "linked.json"
+        try:
+            linked.symlink_to(outside)
+        except OSError as exc:
+            self.skipTest(f"file symlinks are unavailable: {exc}")
+        with self.assertRaisesRegex(R.ArtifactError, "links or reparse points"):
+            R.ingest([linked], self.memdir)
+        self.assertFalse((self.memdir / M.RESEARCH_ARTIFACTS).exists())
+
+    def test_nested_symlink_fails_closed_before_store_write(self):
+        outside = self.root / "outside.json"
+        outside.write_text(json.dumps(_fixtures()[0]), encoding="utf-8")
+        linked = self.inputs / "linked.json"
+        try:
+            linked.symlink_to(outside)
+        except OSError as exc:
+            self.skipTest(f"file symlinks are unavailable: {exc}")
+        with self.assertRaisesRegex(R.ArtifactError, "links or reparse points"):
+            R.ingest([self.inputs], self.memdir)
+        self.assertFalse((self.memdir / M.RESEARCH_ARTIFACTS).exists())
+
+    def test_ancestor_directory_link_is_rejected_before_read(self):
+        outside = self.root / "outside-tree"
+        child = outside / "child"
+        child.mkdir(parents=True)
+        (child / "artifact.json").write_text(
+            json.dumps(_fixtures()[0]), encoding="utf-8"
+        )
+        linked = self.root / "selected-tree"
+        _make_directory_link(self, linked, outside)
+
+        with self.assertRaisesRegex(R.ArtifactError, "links or reparse points"):
+            R.ingest([linked / "child"], self.memdir)
+        self.assertFalse((self.memdir / M.RESEARCH_ARTIFACTS).exists())
+
+    def test_regular_file_replacement_after_discovery_fails_closed(self):
+        source = self.inputs / "artifact-0.json"
+        replacement = self.root / "replacement.json"
+        replacement.write_text(json.dumps(_fixtures()[1]), encoding="utf-8")
+        original_input_files = R._input_files
+
+        def replace_after_discovery(inputs, target):
+            selected = original_input_files(inputs, target)
+            backup = self.root / "original.json"
+            source.replace(backup)
+            replacement.replace(source)
+            return selected
+
+        with mock.patch.object(R, "_input_files", side_effect=replace_after_discovery):
+            with self.assertRaisesRegex(R.ArtifactError, "changed before"):
+                R.ingest([source], self.memdir)
+        self.assertFalse((self.memdir / M.RESEARCH_ARTIFACTS).exists())
+
+    def test_destination_directory_link_is_rejected_without_external_write(self):
+        outside = self.root / "outside-memory"
+        outside.mkdir()
+        linked_memory = self.root / "selected-memory"
+        _make_directory_link(self, linked_memory, outside)
+
+        with self.assertRaisesRegex(R.ArtifactError, "busy or its lock is unsafe"):
+            R.ingest([self.inputs / "artifact-0.json"], linked_memory)
+        self.assertFalse((outside / M.RESEARCH_ARTIFACTS).exists())
+        self.assertFalse((outside / f".{M.RESEARCH_ARTIFACTS}.lock").exists())
+
+    def test_existing_target_and_lock_links_are_rejected_without_external_write(self):
+        self.memdir.mkdir()
+        source = self.inputs / "artifact-0.json"
+        for leaf in (M.RESEARCH_ARTIFACTS, f".{M.RESEARCH_ARTIFACTS}.lock"):
+            with self.subTest(leaf=leaf):
+                outside = self.root / f"outside-{leaf.replace('.', '-')}"
+                outside.mkdir()
+                linked = self.memdir / leaf
+                _make_directory_link(self, linked, outside)
+                try:
+                    with self.assertRaises(R.ArtifactError):
+                        R.ingest([source], self.memdir)
+                    self.assertEqual(list(outside.iterdir()), [])
+                finally:
+                    _remove_directory_link(linked)
+
+    def test_target_replacement_before_publish_fails_closed(self):
+        R.ingest([self.inputs / "artifact-0.json"], self.memdir)
+        target = self.memdir / M.RESEARCH_ARTIFACTS
+        original_load = R._load_existing
+
+        def replace_target(path):
+            records = original_load(path)
+            target.unlink()
+            target.mkdir()
+            return records
+
+        with mock.patch.object(R, "_load_existing", side_effect=replace_target):
+            with self.assertRaises(R.ArtifactError):
+                R.ingest([self.inputs / "artifact-1.json"], self.memdir)
+        self.assertTrue(target.is_dir())
+        self.assertEqual(list(target.iterdir()), [])
+
+    def test_windows_reparse_attribute_is_treated_as_unsafe(self):
+        reparse = getattr(R.stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)
+        info = SimpleNamespace(st_mode=R.stat.S_IFREG, st_file_attributes=reparse)
+        with mock.patch.object(R.stat, "FILE_ATTRIBUTE_REPARSE_POINT", reparse, create=True):
+            self.assertTrue(R._is_link_or_reparse(info))
+
+    def test_read_error_does_not_echo_private_source_path(self):
+        source = self.inputs / "artifact-0.json"
+        private_path = "\\".join(("C:", "Users", "Researcher", "private.json"))
+        private_detail = f"access denied: {private_path}"
+        with mock.patch.object(R, "_read_safe_bytes", side_effect=OSError(private_detail)):
+            with self.assertRaises(R.ArtifactError) as raised:
+                R.ingest([source], self.memdir)
+        self.assertNotIn("C:\\Users", str(raised.exception))
+        self.assertNotIn("private.json", str(raised.exception))
+        self.assertFalse((self.memdir / M.RESEARCH_ARTIFACTS).exists())
+
+    def test_file_uri_is_redacted_before_persistence_and_search(self):
+        R.ingest([self.inputs], self.memdir)
+        text = (self.memdir / M.RESEARCH_ARTIFACTS).read_text(encoding="utf-8")
+        self.assertNotIn("file://", text)
+        self.assertNotIn("hidden-paper.pdf", text)
+        self.assertIn("[REDACTED:local-path]", text)
+
+        (self.root / "inbox").mkdir()
+        database = self.root / "index.db"
+        S.build(self.root / "inbox", db=database)
+        self.assertEqual(S.query("hidden-paper.pdf", db=database, exact=True), [])
+
+    def test_embedded_file_uri_with_spaces_is_fully_redacted(self):
+        document = _fixtures()[0]
+        document["metadata"]["local_uri"] = (
+            "See (file:///C:/Users/Demo/Research and Development/"
+            "Data with Spaces/plan.md) and continue\n"
+            "Also see file:///srv/private/hidden paper.pdf\nNext line"
+        )
+        document["metadata"]["table_uri"] = (
+            "| file:///srv/private/table plan.md | note |"
+        )
+        document["metadata"]["public_url"] = "https://example.org/C:/docs/readme"
+
+        record = R.normalize_artifact(document)
+        text = json.dumps(record, ensure_ascii=False)
+
+        self.assertNotIn("file:", text.casefold())
+        self.assertNotIn("Research and Development", text)
+        self.assertNotIn("Data with Spaces", text)
+        self.assertNotIn("plan.md", text)
+        self.assertNotIn("hidden paper.pdf", text)
+        self.assertNotIn("table plan.md", text)
+        self.assertIn("[REDACTED:local-path]", text)
+        self.assertIn("https://example.org/C:/docs/readme", text)
+
+    def test_store_read_error_does_not_echo_private_destination_path(self):
+        target = self.memdir / M.RESEARCH_ARTIFACTS
+        target.mkdir(parents=True)
+        with self.assertRaises(R.ArtifactError) as raised:
+            R.ingest([self.inputs / "artifact-0.json"], self.memdir)
+        self.assertEqual(str(raised.exception), "existing memory store is unreadable")
+        self.assertNotIn(str(self.memdir), str(raised.exception))
+
+    def test_store_write_error_does_not_echo_private_destination_path(self):
+        private_detail = rf"access denied: {self.memdir}\research-artifacts.json"
+        with mock.patch.object(R, "_atomic_write_json", side_effect=OSError(private_detail)):
+            with self.assertRaises(R.ArtifactError) as raised:
+                R.ingest([self.inputs / "artifact-0.json"], self.memdir)
+        self.assertEqual(
+            str(raised.exception),
+            "research artifact store could not be read or updated",
+        )
+        self.assertNotIn(str(self.memdir), str(raised.exception))
+
+
+class TestResearchArtifactSearch(unittest.TestCase):
+    def test_search_indexes_all_research_artifact_types(self):
+        temp = tempfile.TemporaryDirectory()
+        self.addCleanup(temp.cleanup)
+        root = Path(temp.name)
+        inputs = root / "inputs"
+        inputs.mkdir()
+        for index, document in enumerate(_fixtures()):
+            (inputs / f"{index}.json").write_text(json.dumps(document), encoding="utf-8")
+        R.ingest([inputs], root / "memory")
+        (root / "inbox").mkdir()
+        db = root / "index.db"
+        S.build(root / "inbox", db=db)
+
+        expected = {
+            "microthermometry": "parsed-paper",
+            "geothermometry": "reading-note",
+            "Bibliometric": "review",
+            "lineage anchor": "lineage-graph",
+        }
+        for query, artifact_type in expected.items():
+            with self.subTest(query=query):
+                hits = S.query(query, db=db, exact=True)
+                self.assertTrue(any(hit["source"] == artifact_type for hit in hits), hits)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)
diff --git a/tests/test_status_cli.py b/tests/test_status_cli.py
index 185f0b7..b262904 100644
--- a/tests/test_status_cli.py
+++ b/tests/test_status_cli.py
@@ -24,7 +24,7 @@ def test_version_is_exact_and_does_not_resolve_data_root(self):
                     status.main(["--version"])
 
         self.assertEqual(raised.exception.code, 0)
-        self.assertEqual(stdout.getvalue(), "0.17.0\n")
+        self.assertEqual(stdout.getvalue(), "0.18.0\n")
         self.assertEqual(stderr.getvalue(), "")
         data_root.assert_not_called()
 
diff --git a/tests/test_synclayer_pull.py b/tests/test_synclayer_pull.py
index 9865d99..fb0a54b 100644
--- a/tests/test_synclayer_pull.py
+++ b/tests/test_synclayer_pull.py
@@ -17,6 +17,7 @@
 ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(ROOT))
 
+from provenance import memory as M  # noqa: E402
 from provenance.synclayer import approvals, enqueue, paths, progress_log, pull  # noqa: E402
 from provenance.synclayer import scan_agents, worker  # noqa: E402
 
@@ -121,7 +122,7 @@ def test_capabilities_json_is_stable_and_versioned(self):
         self.assertEqual(report["format_version"], 1)
         self.assertEqual(report["operation"], "pull.capabilities")
         self.assertEqual(
-            report["generated_by"], {"name": "provenance", "version": "0.17.0"}
+            report["generated_by"], {"name": "provenance", "version": "0.18.0"}
         )
         self.assertEqual(report["action_required"], [])
         self.assertEqual(
@@ -175,7 +176,7 @@ def test_pull_json_stdout_is_one_structured_object(self):
             },
         )
         self.assertEqual(
-            report["generated_by"], {"name": "provenance", "version": "0.17.0"}
+            report["generated_by"], {"name": "provenance", "version": "0.18.0"}
         )
 
     def test_preview_uses_env_fallback_and_writes_no_authoritative_data(self):
@@ -311,6 +312,10 @@ def test_run_applies_existing_fill_then_only_checked_approval(self):
             note.read_text(encoding="utf-8").count(progress_log.entry_marker(summary_id)),
             1,
         )
+        self.assertIn(
+            "implemented the enqueue hook + cwd->project_id resolver",
+            M.load_projects(self.root)["alpha"]["body"],
+        )
 
         second = pull.run_pull(
             workspace=self.workspace, provenance_home=self.root, run=True
@@ -333,6 +338,10 @@ def test_run_applies_existing_fill_then_only_checked_approval(self):
         self.assertEqual(third["summary"]["approved"], 1)
         self.assertEqual(third["summary"]["pending_approval"], 0)
         self.assertIn("status: active", note.read_text(encoding="utf-8"))
+        self.assertIn(
+            progress_log.entry_marker(f"{summary_id}#approved"),
+            M.load_projects(self.root)["alpha"]["body"],
+        )
         self.assertNotIn(summary_id, json.dumps(third, ensure_ascii=False))
 
     def test_malformed_jsonl_blocks_before_any_authoritative_write(self):
diff --git a/uv.lock b/uv.lock
index 08660b5..f966e50 100644
--- a/uv.lock
+++ b/uv.lock
@@ -4,5 +4,5 @@ requires-python = ">=3.9"
 
 [[package]]
 name = "provenance"
-version = "0.17.0"
+version = "0.18.0"
 source = { editable = "." }