fix: harden semantic dream apply routing

Author: deeflect

src/dory_core/semantic_write.py

@@ -89,14 +89,20 @@ def __init__(
         resolver_client: OpenRouterClient | None = None,
     ) -> None:
         self.root = Path(root)
-        self.writer = WriteEngine(root=self.root, index_root=index_root, embedder=embedder)
+        settings = DorySettings()
+        self.writer = WriteEngine(
+            root=self.root,
+            max_write_bytes=max(settings.max_write_bytes, 256_000),
+            index_root=index_root,
+            embedder=embedder,
+        )
         self.registry = EntityRegistry(self.root / ".dory" / "entity-registry.db")
         self.claim_store = ClaimStore(self.root / ".dory" / "claim-store.db")
         resolved_client = (
             resolver_client
             if resolver_client is not None
             else build_openrouter_client(
-                DorySettings(),
+                settings,
                 purpose="maintenance",
             )
         )
@@ -670,6 +676,8 @@ def build_semantic_write_plan(
 ) -> SemanticWritePlan:
     resolver = resolver or SubjectResolver(root)
     match = resolver.resolve(req.subject, scope=req.scope)
+    if match is None or _should_create_new_explicit_dream_subject(match, req):
+        match = _new_subject_match_from_explicit_scope(req)
     if match is None:
         raise ValueError(f"could not resolve semantic subject: {req.subject}")
 
@@ -699,6 +707,38 @@ def build_semantic_write_plan(
     )
 
 
+def _new_subject_match_from_explicit_scope(req: MemoryWriteReq) -> SubjectMatch | None:
+    if req.scope not in {"project", "concept", "decision"}:
+        return None
+    slug = normalize_migration_slug(req.subject)
+    if not slug:
+        return None
+    subject_ref = f"{req.scope}:{slug}"
+    return SubjectMatch(
+        subject_ref=subject_ref,
+        family=req.scope,
+        title=canonical_title_from_subject(subject_ref),
+        target_path=canonical_target_for_subject(subject_ref),
+        matched_by="explicit_scope",
+        confidence="high",
+    )
+
+
+def _should_create_new_explicit_dream_subject(match: SubjectMatch, req: MemoryWriteReq) -> bool:
+    if req.scope not in {"project", "concept", "decision"}:
+        return False
+    source = req.source or ""
+    if "/digests/" not in source and "/inbox/distilled/" not in source:
+        return False
+    slug = normalize_migration_slug(req.subject)
+    if not slug:
+        return False
+    requested_subject_ref = f"{req.scope}:{slug}"
+    if match.subject_ref == requested_subject_ref:
+        return False
+    return match.matched_by in {"alias", "llm"}
+
+
 def _route_target(match: SubjectMatch, req: MemoryWriteReq) -> tuple[str, str, str]:
     if match.family == "core":
         return match.subject_ref, "core", match.target_path

tests/integration/core/test_semantic_write_flow.py

@@ -132,6 +132,33 @@ def test_semantic_write_flow_creates_structured_decision_pages(tmp_path: Path) -
     assert "## Evidence" in decision.body
 
 
+def test_semantic_write_engine_allows_large_canonical_rewrites(tmp_path: Path) -> None:
+    root = tmp_path / "corpus"
+    (root / "projects" / "dory").mkdir(parents=True)
+    project_path = root / "projects" / "dory" / "state.md"
+    project_path.write_text(
+        "---\ntitle: Dory\naliases: []\n---\n# Dory\n\n## Current State\n\n"
+        + ("Existing canonical context.\n" * 700),
+        encoding="utf-8",
+    )
+
+    engine = SemanticWriteEngine(root)
+    response = engine.write(
+        MemoryWriteReq(
+            action="write",
+            kind="state",
+            subject="dory",
+            content="Dory can update large canonical pages.",
+            scope="project",
+            allow_canonical=True,
+        )
+    )
+
+    assert response.resolved is True
+    assert response.result == "written"
+    assert "Dory can update large canonical pages." in project_path.read_text(encoding="utf-8")
+
+
 def test_semantic_write_dry_run_reports_canonical_target_without_persisting(tmp_path: Path) -> None:
     root = tmp_path / "corpus"
     (root / "projects" / "dory").mkdir(parents=True)
@@ -193,7 +220,7 @@ def test_semantic_write_dry_run_large_canonical_target_still_reports_route(tmp_p
     assert response.target_path == "projects/dory/state.md"
     assert response.message is not None
     assert response.message.startswith("CANONICAL TARGET projects/dory/state.md")
-    assert "rendered target exceeds preview write-size limit" in response.message
+    assert "dry_run: would_replace" in response.message
     assert project_path.read_text(encoding="utf-8") == before
 
 

tests/unit/test_semantic_write.py

@@ -2,7 +2,7 @@
 
 from pathlib import Path
 
-from dory_core.semantic_write import SubjectResolver, build_semantic_write_plan
+from dory_core.semantic_write import SubjectMatch, SubjectResolver, build_semantic_write_plan
 from dory_core.types import MemoryWriteReq, MemoryWriteResp
 
 
@@ -124,3 +124,97 @@ def test_build_semantic_write_plan_routes_to_canonical_targets(tmp_path: Path) -
     assert core_plan.subject_ref == "core:user"
     assert core_plan.target_path == "core/user.md"
     assert core_plan.resolved_mode == "replace"
+
+
+def test_build_semantic_write_plan_creates_new_project_from_explicit_scope(tmp_path: Path) -> None:
+    plan = build_semantic_write_plan(
+        tmp_path,
+        MemoryWriteReq(
+            action="write",
+            kind="state",
+            subject="Open Privacy Filter",
+            content="Open Privacy Filter is active.",
+            scope="project",
+        ),
+    )
+
+    assert plan.subject_ref == "project:open-privacy-filter"
+    assert plan.target_subject_ref == "project:open-privacy-filter"
+    assert plan.target_path == "projects/open-privacy-filter/state.md"
+    assert plan.matched_by == "explicit_scope"
+    assert plan.target_exists is False
+
+
+def test_build_semantic_write_plan_does_not_create_people_from_explicit_scope(tmp_path: Path) -> None:
+    try:
+        build_semantic_write_plan(
+            tmp_path,
+            MemoryWriteReq(
+                action="write",
+                kind="preference",
+                subject="active",
+                content="Bad scoped proposal should not create a person.",
+                scope="person",
+            ),
+        )
+    except ValueError as err:
+        assert "could not resolve semantic subject: active" in str(err)
+    else:
+        raise AssertionError("expected unresolved person subject to be rejected")
+
+
+def test_build_semantic_write_plan_creates_new_dream_project_over_alias_match(tmp_path: Path) -> None:
+    class _AliasResolver:
+        def resolve(self, subject: str, *, scope: str | None = None) -> SubjectMatch | None:
+            return SubjectMatch(
+                subject_ref="project:privacy-filter-lab",
+                family="project",
+                title="Privacy Filter Lab",
+                target_path="projects/privacy-filter-lab/state.md",
+                matched_by="alias",
+                confidence="high",
+            )
+
+    plan = build_semantic_write_plan(
+        tmp_path,
+        MemoryWriteReq(
+            action="write",
+            kind="state",
+            subject="open-privacy-filter",
+            content="Open Privacy Filter is active.",
+            scope="project",
+            source="/var/lib/dory/digests/daily/2026-04-23.md",
+        ),
+        resolver=_AliasResolver(),
+    )
+
+    assert plan.subject_ref == "project:open-privacy-filter"
+    assert plan.target_path == "projects/open-privacy-filter/state.md"
+
+
+def test_build_semantic_write_plan_keeps_alias_match_for_non_dream_write(tmp_path: Path) -> None:
+    class _AliasResolver:
+        def resolve(self, subject: str, *, scope: str | None = None) -> SubjectMatch | None:
+            return SubjectMatch(
+                subject_ref="project:privacy-filter-lab",
+                family="project",
+                title="Privacy Filter Lab",
+                target_path="projects/privacy-filter-lab/state.md",
+                matched_by="alias",
+                confidence="high",
+            )
+
+    plan = build_semantic_write_plan(
+        tmp_path,
+        MemoryWriteReq(
+            action="write",
+            kind="state",
+            subject="open-privacy-filter",
+            content="Open Privacy Filter is active.",
+            scope="project",
+        ),
+        resolver=_AliasResolver(),
+    )
+
+    assert plan.subject_ref == "project:privacy-filter-lab"
+    assert plan.target_path == "projects/privacy-filter-lab/state.md"