diff --git a/backend/services/assistant_materials.py b/backend/services/assistant_materials.py
index f1f8960..18fdc79 100644
--- a/backend/services/assistant_materials.py
+++ b/backend/services/assistant_materials.py
@@ -55,18 +55,29 @@ def parse(content, filename):
         raise ValueError("材料为空或超过 10 MB")
     suffix = Path(filename).suffix.lower()
     chunks, warnings = [], []
+    stem = Path(filename).stem
     def add(locator, text, **extra):
         if len(chunks) >= 15000:
             raise ValueError("材料内容超过首版处理上限")
         chunks.append({"locator": locator, "text": str(text)[:16000], **extra})
+    # 类型化定位：与自由字符串 locator 并存（claude_use/T00/04 §5.3）。
+    # **保留 locator** —— enqueue() 用它算证据 ID 的稳定哈希，改它会破坏版本冻结。
+    # document_id / document_version 由 material() 补齐后按契约校验。
+    def pdf(page):
+        return {"kind": "pdf", "page": page, "excerpt": f"第{page}页"}
+    def sheet_loc(sheet, cell_range):
+        return {"kind": "table", "sheet": sheet, "cell_range": cell_range,
+                "excerpt": f"{sheet}!{cell_range}" if sheet else cell_range}
+    def manual(note):
+        return {"kind": "manual", "note": note, "excerpt": note}
     if suffix in (".txt", ".csv"):
         text = content.decode("utf-8-sig", errors="strict")
         if suffix == ".txt":
             for index, line in enumerate(text.splitlines(), 1):
-                add(f"第{index}行", line)
+                add(f"第{index}行", line, source_locator=sheet_loc(stem, f"第{index}行"))
         else:
             for index, row in enumerate(csv.reader(io.StringIO(text)), 1):
-                add(f"第{index}行", " | ".join(row), cells=row)
+                add(f"第{index}行", " | ".join(row), cells=row, source_locator=sheet_loc(stem, f"第{index}行"))
     elif suffix == ".docx":
         with safe_zip(content) as archive:
             root = ET.fromstring(archive.read("word/document.xml"))
@@ -75,12 +86,15 @@ def parse(content, filename):
         for element in root.find("w:body", ns):
             if element.tag.endswith("}p"):
                 para += 1
-                add(f"正文第{para}段", "".join(t.text or "" for t in element.findall(".//w:t", ns)))
+                # 正文段落不是行列可寻址的表格，归 manual（设计的三值枚举里没有第四种）
+                add(f"正文第{para}段", "".join(t.text or "" for t in element.findall(".//w:t", ns)),
+                    source_locator=manual(f"正文第{para}段"))
             elif element.tag.endswith("}tbl"):
                 table += 1
                 for r, row in enumerate(element.findall("w:tr", ns), 1):
                     cells = ["".join(t.text or "" for t in cell.findall(".//w:t", ns)) for cell in row.findall("w:tc", ns)]
-                    add(f"表{table}第{r}行", " | ".join(cells), cells=cells)
+                    add(f"表{table}第{r}行", " | ".join(cells), cells=cells,
+                        source_locator=sheet_loc(f"表{table}", f"第{r}行"))
     elif suffix == ".xlsx":
         ns = {"x": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
         with safe_zip(content) as archive:
@@ -90,6 +104,7 @@ def parse(content, filename):
             rels = {r.get("Id"): r.get("Target") for r in ET.fromstring(archive.read("xl/_rels/workbook.xml.rels"))}
             book = ET.fromstring(archive.read("xl/workbook.xml"))
             for sheet in book.findall("x:sheets/x:sheet", ns):
+                sheet_name = sheet.get("name")
                 target = rels.get(sheet.get("{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"), "")
                 path = target.lstrip("/") if target.startswith("/") else "xl/" + target
                 if ".." in path or path not in archive.namelist():
@@ -102,15 +117,19 @@ def parse(content, filename):
                         text = strings[int(raw)] if kind == "s" and raw else ("".join(t.text or "" for t in cell.findall(".//x:t", ns)) if kind == "inlineStr" else raw)
                         formula = cell.findtext("x:f", namespaces=ns)
                         if formula is not None and raw == "":
-                            warnings.append(sheet.get("name") + "!" + cell.get("r") + "公式无缓存，未计算")
-                        add(sheet.get("name") + "!" + cell.get("r"), text, formula=formula, cached_value=raw if formula is not None else None)
+                            warnings.append(sheet_name + "!" + cell.get("r") + "公式无缓存，未计算")
+                        add(sheet_name + "!" + cell.get("r"), text, formula=formula,
+                            cached_value=raw if formula is not None else None,
+                            source_locator=sheet_loc(sheet_name, cell.get("r")))
                         cells.append(text)
                     if cells:
-                        add(sheet.get("name") + "!第" + row.get("r") + "行", " | ".join(cells), cells=cells)
+                        add(sheet_name + "!第" + row.get("r") + "行", " | ".join(cells), cells=cells,
+                            source_locator=sheet_loc(sheet_name, f"第{row.get('r')}行"))
     elif suffix in (".pdf", ".png", ".jpg", ".jpeg"):
         from api.collateral_agent import extract_pages
         for page in extract_pages(content, filename):
-            add("第" + str(page["number"]) + "页", page.get("text", ""), method=page.get("method"), status=page.get("status"))
+            add("第" + str(page["number"]) + "页", page.get("text", ""), method=page.get("method"),
+                status=page.get("status"), source_locator=pdf(page["number"]))
             if not page.get("text", "").strip():
                 warnings.append("第" + str(page["number"]) + "页未提取到文字，请启用 OCR 或补充可读材料")
     else:
@@ -126,7 +145,23 @@ def material(project_id, filename, content, actor, version=1):
         chunks, warnings = parse(content, filename)
     except (OSError, zipfile.BadZipFile, ET.ParseError, UnicodeError, csv.Error, IndexError, TypeError) as exc:
         raise ValueError("材料内容损坏、编码不支持或格式不匹配，请检查原文件") from exc
-    body = {"id": uid("doc"), "project_id": project_id, "name": filename, "version": version,
+    document_id = uid("doc")
+    # 补齐并**按契约校验**类型化定位：document_id/document_version 只有到这一步才知道，
+    # 而 SourceLocator 要求二者必填。校验失败即视为解析产物不合规（不能落到库里当"定位"）。
+    from pydantic import ValidationError
+    from contracts.prescreen import SourceLocator
+    for index, chunk in enumerate(chunks):
+        locator = chunk.get("source_locator")
+        if locator is None:
+            raise ValueError(f"解析产物缺少类型化定位（第 {index + 1} 段）")
+        try:
+            chunk["source_locator"] = SourceLocator(
+                **locator, document_id=document_id, document_version=version
+            ).model_dump(mode="json")
+        except ValidationError as exc:
+            first = exc.errors()[0]
+            raise ValueError(f"定位不合法（{first['loc'][-1] if first['loc'] else '未知'}）：{first['msg']}") from exc
+    body = {"id": document_id, "project_id": project_id, "name": filename, "version": version,
             "content_hash": sha256(content).hexdigest(), "created_at": now(), "actor_id": actor,
             "size": len(content), "chunks": chunks, "warnings": warnings,
             "status": "partial" if warnings else "parsed", "raw_base64": base64.b64encode(content).decode(), "candidates": []}
diff --git a/backend/test_sourcelocator.py b/backend/test_sourcelocator.py
new file mode 100644
index 0000000..4d696e0
--- /dev/null
+++ b/backend/test_sourcelocator.py
@@ -0,0 +1,210 @@
+"""T03 类型化来源定位测试：parse()/material() 必须产出契约合法的 SourceLocator。
+
+背景：T02 定义了 `contracts.prescreen.SourceLocator`，但当时**没有接上**
+（parse() 只产出自由字符串 locator）。本文件锁定这条链路：
+    parse() 产出 kind + 坐标 → material() 补 document_id/document_version → 契约校验
+
+同时锁定 `locator` 自由字符串**必须保留**：enqueue() 用它算证据 ID 的稳定哈希，
+改动它会破坏 BR-13 的版本冻结。
+"""
+
+import io
+import unittest
+import zipfile
+from pathlib import Path
+
+from contracts.prescreen import SourceLocator
+from services.assistant_materials import material, parse
+
+
+def xlsx_bytes(rows=(("field", "value"), ("facility.amount", "100"))):
+    """构造最小可解析 XLSX（inlineStr，无需 sharedStrings）。"""
+    ns = 'xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"'
+    rel = 'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"'
+    sheet_rows = "".join(
+        "<row r=\"%d\">%s</row>" % (
+            index,
+            "".join(
+                f'<c r="{chr(65 + col)}{index}" t="inlineStr"><is><t>{cell}</t></is></c>'
+                for col, cell in enumerate(row)
+            ),
+        )
+        for index, row in enumerate(rows, 1)
+    )
+    parts = {
+        "xl/workbook.xml": f'<workbook {ns}><sheets><sheet name="Sheet1" sheetId="1" r:id="rId1" {rel}/></sheets></workbook>',
+        "xl/_rels/workbook.xml.rels": '<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
+                                      '<Relationship Id="rId1" Target="worksheets/sheet1.xml"/></Relationships>',
+        "xl/worksheets/sheet1.xml": f'<worksheet {ns}><sheetData>{sheet_rows}</sheetData></worksheet>',
+    }
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w") as archive:
+        for name, body in parts.items():
+            archive.writestr(name, body)
+    return buffer.getvalue()
+
+
+def docx_bytes():
+    """构造最小可解析 DOCX：一段正文 + 一张表。"""
+    ns = 'xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main"'
+    body = (
+        f'<w:p><w:r><w:t>正文第一段</w:t></w:r></w:p>'
+        f'<w:tbl><w:tr><w:tc><w:p><w:r><w:t>甲</w:t></w:r></w:p></w:tc>'
+        f'<w:tc><w:p><w:r><w:t>乙</w:t></w:r></w:p></w:tc></w:tr></w:tbl>'
+    )
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w") as archive:
+        archive.writestr("word/document.xml", f"<w:document {ns}><w:body>{body}</w:body></w:document>")
+    return buffer.getvalue()
+
+
+class TypedLocatorTests(unittest.TestCase):
+    def locators(self, content, filename):
+        document = material("DEMO-A", filename, content, "tester")
+        return document, [chunk["source_locator"] for chunk in document["chunks"]]
+
+    def test_csv_rows_become_table_locators(self):
+        content = "field,value\nfacility.amount,100\n".encode("utf-8-sig")
+        document, locators = self.locators(content, "流水.csv")
+        first = locators[0]
+        self.assertEqual(first["kind"], "table")
+        self.assertEqual(first["sheet"], "流水")       # 文件名去扩展名
+        self.assertEqual(first["cell_range"], "第1行")
+        self.assertEqual(first["document_id"], document["id"])
+        self.assertEqual(first["document_version"], 1)
+
+    def test_txt_rows_become_table_locators(self):
+        content = "第一行\n第二行\n".encode("utf-8-sig")
+        _, locators = self.locators(content, "说明.txt")
+        self.assertEqual(locators[1]["cell_range"], "第2行")
+        self.assertEqual(locators[1]["kind"], "table")
+
+    def test_xlsx_cells_and_rows_become_table_locators(self):
+        _, locators = self.locators(xlsx_bytes(), "台账.xlsx")
+        kinds = {item["kind"] for item in locators}
+        self.assertEqual(kinds, {"table"})
+        # 单元格级定位：A1 必须是 sheet + cell_range，而不是行号
+        cell = next(item for item in locators if item["cell_range"] == "A1")
+        self.assertEqual(cell["sheet"], "Sheet1")
+        # 行级定位也保留
+        self.assertTrue(any(item["cell_range"].endswith("行") for item in locators))
+
+    def test_docx_paragraph_is_manual_and_table_is_table(self):
+        _, locators = self.locators(docx_bytes(), "说明.docx")
+        by_kind = {}
+        for item in locators:
+            by_kind.setdefault(item["kind"], []).append(item)
+        self.assertIn("manual", by_kind, "DOCX 正文段应归 manual（三值枚举里没有段落子类型）")
+        self.assertIn("table", by_kind, "DOCX 表格应归 table")
+        self.assertEqual(by_kind["manual"][0]["note"], "正文第1段")
+        self.assertEqual(by_kind["table"][0]["sheet"], "表1")
+
+    def test_every_locator_validates_against_the_contract(self):
+        # material() 已按契约校验；这里再独立复核一遍，防止校验被绕过
+        for content, name in [(xlsx_bytes(), "台账.xlsx"), (docx_bytes(), "说明.docx"),
+                              ("a,b\n".encode("utf-8-sig"), "t.csv")]:
+            with self.subTest(name=name):
+                _, locators = self.locators(content, name)
+                self.assertTrue(locators)
+                for item in locators:
+                    SourceLocator(**item)   # 不合法会抛 ValidationError
+
+    def test_free_string_locator_is_preserved(self):
+        """locator 必须保留：enqueue() 依赖它计算证据 ID 的稳定哈希（BR-13）。"""
+        content = "field,value\nfacility.amount,100\n".encode("utf-8-sig")
+        document = material("DEMO-A", "流水.csv", content, "tester")
+        strings = [chunk["locator"] for chunk in document["chunks"]]
+        self.assertIn("第1行", strings)
+        self.assertIn("第2行", strings)
+        # 每个 chunk 同时有两种形态
+        for chunk in document["chunks"]:
+            self.assertIn("locator", chunk)
+            self.assertIn("source_locator", chunk)
+            self.assertIsInstance(chunk["locator"], str)
+
+    def test_parse_without_material_fillup_has_no_document_id(self):
+        """parse() 本身不知道 document_id——这正是补齐放在 material() 的原因。"""
+        chunks, _ = parse("a\n".encode("utf-8-sig"), "t.txt")
+        raw = chunks[0]["source_locator"]
+        self.assertNotIn("document_id", raw)
+        self.assertEqual(raw["kind"], "table")
+
+    def test_document_version_is_recorded_not_assumed(self):
+        content = "a,b\n".encode("utf-8-sig")
+        document = material("DEMO-A", "v.csv", content, "tester", version=7)
+        self.assertEqual(document["version"], 7)
+        self.assertEqual(document["chunks"][0]["source_locator"]["document_version"], 7)
+
+    def test_locator_cannot_cross_documents(self):
+        """两个文档的定位互不串用：document_id 是各自的。"""
+        a = material("DEMO-A", "a.csv", "x,y\n".encode("utf-8-sig"), "tester")
+        b = material("DEMO-A", "b.csv", "x,y\n".encode("utf-8-sig"), "tester")
+        self.assertNotEqual(a["id"], b["id"])
+        self.assertEqual(a["chunks"][0]["source_locator"]["document_id"], a["id"])
+        self.assertEqual(b["chunks"][0]["source_locator"]["document_id"], b["id"])
+        # 文档 a 的定位不可能出现在文档 b 里
+        self.assertNotIn(a["id"], [c["source_locator"]["document_id"] for c in b["chunks"]])
+
+    def test_missing_typed_locator_is_rejected(self):
+        """防御性断言：若某条解析路径漏了 source_locator，material() 必须失败而不是静默产半成品。"""
+        import services.assistant_materials as materials
+        from unittest.mock import patch
+        original = materials.parse
+
+        def stripped(content, filename):
+            chunks, warnings = original(content, filename)
+            for chunk in chunks:
+                chunk.pop("source_locator", None)
+            return chunks, warnings
+
+        with patch.object(materials, "parse", stripped):
+            with self.assertRaises(ValueError) as ctx:
+                materials.material("DEMO-A", "t.csv", "a,b\n".encode("utf-8-sig"), "tester")
+            self.assertIn("类型化定位", str(ctx.exception))
+
+
+class ReferenceValidationTests(unittest.TestCase):
+    """AC-08：无效或跨项目引用必须被拒，且不暴露无权原文。"""
+
+    def setUp(self):
+        import tempfile
+        from services.project_assistant import ProjectAssistantService
+        self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+        self.service = ProjectAssistantService(Path(self.temp.name) / "assistant.sqlite3")
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def test_candidate_from_another_project_is_rejected(self):
+        payload = "field,value,currency,unit,period,scope\nfacility.amount,100,CNY,万元,2025,合并\n".encode("utf-8-sig")
+        document = self.service.upload("DEMO-A", "流水.csv", payload, "demo-manager")
+        self.assertTrue(document["candidates"], "解析应产出候选")
+        candidate_id = document["candidates"][0]["id"]
+
+        with self.assertRaises(ValueError) as ctx:
+            self.service.supplement("DEMO-B", "demo-manager", {}, [],
+                                    self.service.get_project("DEMO-B")["revision"],
+                                    "跨项目引用测试", candidate_ids=[candidate_id])
+        self.assertIn("不属于项目", str(ctx.exception))
+
+    def test_candidate_reference_is_project_scoped(self):
+        """同项目可用、跨项目不可用——证明拒绝来自项目范围而不是候选不存在。"""
+        payload = "field,value,currency,unit,period,scope\nfacility.amount,100,CNY,万元,2025,合并\n".encode("utf-8-sig")
+        document = self.service.upload("DEMO-A", "流水.csv", payload, "demo-manager")
+        candidate_id = document["candidates"][0]["id"]
+
+        project = self.service.get_project("DEMO-A")
+        result = self.service.supplement("DEMO-A", "demo-manager", {}, [], project["revision"],
+                                         "同项目确认来源", candidate_ids=[candidate_id])
+        self.assertEqual(result["project_id"], "DEMO-A")
+        self.assertIn("facility.amount", result["changed_fields"])
+
+    def test_unknown_candidate_id_is_also_rejected(self):
+        project = self.service.get_project("DEMO-A")
+        with self.assertRaises(ValueError):
+            self.service.supplement("DEMO-A", "demo-manager", {}, [], project["revision"],
+                                    "伪造引用", candidate_ids=["candidate-does-not-exist"])
+
+
+if __name__ == "__main__":
+    unittest.main()