feat: wire FR-013 predicate/subject_scope/retrieval_context_hash into POST /memory/write (DASH-STORY-026)

Implements docs/phase-1.5-api/fr013-predicate-schema-design.md v7.1
(IMPLEMENTATION-READY, 7 review rounds) exactly, per the Sprint 4
planning bundle (docs/phase-7-routing/sprint4_*, Phase C gate PASS).
Zone 3 (Semantic) and Zone 5 (Entity) writes are now reachable through
the live API, closing GitHub #23.

openapi.yaml: adds predicate/subject_scope to WriteMemoryRequest.content,
retrieval_context_hash (pattern ^[0-9a-f]{64}$) to provenance, widens
WriteReceipt.status to [accepted, partial] with zone5_written/
zone3_edges_written/zone3_edges_failed, documents the new 207 response.

schemas.py: matching Pydantic model changes.

app.py's _do_write: replaces the unconditional 422
ZONE_3_5_PREDICATE_UNRESOLVABLE rejection with real routing (Section
4.2 steps 0-7) -- a hard gate so ordinary Zone 1/2/4 writes never enter
the new path, zone_hint normalized via the existing _WIRE_TO_DOMAIN_ZONE
map (never a bare string, unlike GitHub #25's separate, still-open bug),
CandidateFact/classify() reused unchanged, per-edge provenance_id never
shared, 400 INVALID_REQUEST for domain-constructor ValueErrors (mirrors
assemble_context's existing precedent), 207 Multi-Status for a partial
Zone 3 failure after a successful Zone 5 write, 503 for a Zone 5 failure.

tests/test_api_wire_layer_dash023.py: updates the one pre-story test
that asserted the now-removed 422 behavior to assert the new 400.

New tests/integration/test_api_real_postgres_e2e_dash026_zone3_5_routing.py:
23 real HTTP-level tests against a real docker-compose Postgres stack,
covering the design doc's full 14-case matrix.

Dev->QA->Review (k=2 retry): Review attempt 1 REJECTED on one real
finding (FR-013 cited as "verbatim" but paraphrased in two docstrings);
fixed; attempt 2 APPROVED after full independent re-verification
(real pytest run, real SRS.md text diff, real GitHub issue reads).

Full suite: 1610 passed, 0 failed. 6 of 23 new real-Postgres tests fail
due to a genuinely separate, pre-existing bug (SqlProvenanceRepository
crashing on a no-RETURNING INSERT), disclosed as GitHub #26, correctly
not fixed here (outside this story's file cluster).

Closes #23

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: piyushmakhija28

docs/phase-1.5-api/openapi.yaml

@@ -332,6 +332,20 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/WriteReceipt'
+        '207':
+          description: >-
+            FR-013 design Section 4.2 step 6 -- Zone 5 (Entity) write
+            succeeded but one or more Zone 3 (Semantic) edge writes failed
+            partway through. `WriteReceipt.status` is "partial",
+            `zone5_written` is true, and `zone3_edges_failed` names the
+            edges that did not commit. No rollback of the already-durable
+            Zone 5 write or any already-committed Zone 3 edges is
+            attempted (fail-fast-plus-honest-partial-reporting; true
+            cross-zone atomicity is explicitly out of scope, AR1-S3-G3).
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/WriteReceipt'
         '400':
           $ref: '#/components/responses/BadRequest'
         '401':
@@ -1414,6 +1428,19 @@ components:
         importance: { type: number, minimum: 0, maximum: 1, nullable: true }
         purpose: { type: string, description: DPDP purpose-limitation tag (DPDP-1) }
         subject_id: { type: string, nullable: true, description: Data-subject identifier for the DPDP erasure/export index (ADR-006) }
+        retrieval_context_hash:
+          type: string
+          nullable: true
+          pattern: '^[0-9a-f]{64}$'
+          description: >-
+            SHA-256 hex of the query/task this write was made under,
+            pre-hashed by the caller (this host never receives raw
+            retrieval-context text; matches domain.write_gate.WriteRequest.
+            retrieval_context_hash's own established semantics, FR-013
+            design Section 3/4.2 step 5). Required whenever a Zone 3/5
+            write is intended (content.entity_refs non-empty, or
+            content.zone_hint normalizes to Zone 3); optional and unread
+            for Zone 1/2/4 writes, which are unaffected.
 
     WriteMemoryRequest:
       type: object
@@ -1446,15 +1473,49 @@ components:
                 Section 3.4: when a single-subject, single-object fact should
                 ALSO get a Zone 3 SemanticEdge for reverse traversal, on top of
                 its mandatory Zone 5 attribute write. Default false.
+            predicate:
+              type: string
+              nullable: true
+              description: >-
+                CandidateFact.predicate (HLD 3.4). A single fact-level
+                relation/attribute name applying to the whole write,
+                regardless of how many subject/object pairs classify()
+                expands it into. Required whenever entity_refs contains at
+                least one subject-role entry (FR-013 design Section 4.2
+                step 5b: 400 INVALID_REQUEST if missing). Ignored when
+                entity_refs has no subject-role entry.
+            subject_scope:
+              type: string
+              nullable: true
+              description: >-
+                CandidateFact.subject_scope / GeneralFact.subject_scope
+                (HLD 3.4, the no-subject-entity_refs branch only).
+                Required whenever entity_refs is empty AND the caller
+                intends a GeneralFact write (FR-013 design Section 4.2 step
+                5b: 400 INVALID_REQUEST if missing/blank in that case).
         provenance: { $ref: '#/components/schemas/ProvenanceWriteBlock' }
 
     WriteReceipt:
       type: object
       required: [write_id, status, accepted_at]
       properties:
         write_id: { type: string, format: uuid }
-        status: { type: string, enum: [accepted] }
+        status: { type: string, enum: [accepted, partial] }
         accepted_at: { type: string, format: date-time }
+        zone5_written:
+          type: boolean
+          default: false
+          description: FR-013 design Section 4.2 step 6 -- true when this write's Zone 5 (Entity) attribute write committed.
+        zone3_edges_written:
+          type: array
+          items: { type: string }
+          default: []
+          description: item_ids (subject:predicate:object) of Zone 3 SemanticEdges that committed.
+        zone3_edges_failed:
+          type: array
+          items: { type: string }
+          default: []
+          description: item_ids of Zone 3 SemanticEdges that were attempted and failed (207 Multi-Status case).
 
     BatchWriteReceiptItem:
       type: object

src/dashanan/api/app.py

@@ -23,25 +23,26 @@
 itself as closing HLD Section 7.4's full "cascades across all 8 zones"
 promise (must-not-deviate item 3) until DASH-STORY-025 lands.
 
-Zone 3/5 write-path gap (flagged, never fabricated): openapi.yaml's own
-`WriteMemoryRequest.content` shape (locked; must-not-deviate item 2)
-carries `entity_refs` but no `predicate` field, while
-`domain.semantic_memory.SemanticEdge`/`GeneralFact` both require one
-(HLD Section 3.4's `EntityOwnershipSpecification` classification gate).
-This handler cannot fabricate a predicate value, so an `entity_refs`
--bearing write is rejected `422 UNPROCESSABLE_ENTITY` with an explicit
-`ZONE_3_5_PREDICATE_UNRESOLVABLE` error code -- the operationId still
-resolves to a real, callable handler (AC-023-1); it correctly declines a
-request its own wire contract cannot faithfully route, exactly the class
-of response openapi.yaml's own `422 UnprocessableEntity` documents.
+Zone 3/5 write path (DASH-STORY-026, traces to FR-013, closes GitHub
+#23): `WriteMemoryRequest.content` now carries `predicate`/
+`subject_scope`, and `WriteMemoryRequest.provenance` carries
+`retrieval_context_hash` (docs/phase-1.5-api/fr013-predicate-schema-
+design.md v7, Section 3). `_do_write`'s routing gate (step 0 below)
+sends any write with a non-empty `content.entity_refs`, or a
+`content.zone_hint` that normalizes to Zone 3 (Semantic), through
+`entity_ownership_specification.classify()` and the FR-013-swept Zone
+3/5 repositories (`AppContext.semantic_repository`/
+`conflict_aware_entity_repository`) -- every other write (the ordinary
+Zone 1/2/4 case) falls through to the unchanged branch below, unaffected
+by this story (AC-026-DEV-2).
 """
 
 import logging
 import uuid
 from dataclasses import asdict as _dataclass_asdict
 from collections.abc import Callable
 from datetime import UTC, datetime
-from typing import Annotated
+from typing import Annotated, Literal
 
 from fastapi import Depends, FastAPI, Header, HTTPException, Response
 from fastapi.responses import JSONResponse
@@ -57,7 +58,15 @@
     verify_bearer_token,
 )
 from dashanan.application.context_assembly_request import ContextAssemblyRequest
+from dashanan.application.conflict_aware_zone_writes import _zone3_edge_item_id
+from dashanan.domain.entity_ownership_specification import (
+    CandidateFact,
+    GeneralFactRouting,
+    SemanticEdgeRouting,
+    classify,
+)
 from dashanan.domain.exceptions import ZoneRepositoryError
+from dashanan.domain.provenance_record import SourceType
 from dashanan.domain.subject_erasure import SubjectErasureRequest
 from dashanan.domain.zone import ZoneId
 from dashanan.infrastructure.composition_root import (
@@ -373,21 +382,160 @@ def batch_get_items(
     def _resolve_source_type(body: sc.WriteMemoryRequest) -> str | None:
         return body.provenance.source_type
 
+    def _zone3_5_routing_gate_triggered(body: sc.WriteMemoryRequest) -> bool:
+        """FR-013 design Section 4.2 step 0's normalization rule.
+
+        Never compares `content.zone_hint` against a bare string -- uses
+        the existing `_WIRE_TO_DOMAIN_ZONE` map against `ZoneId.SEMANTIC`,
+        exactly as GitHub #25's disclosed, pre-existing `zone_hint ==
+        "procedural"`/`"episodic"` bare-string comparisons must NOT be
+        repeated (design doc Section 4.2 step 0, "Separate, pre-existing
+        defect... NOT fixed by this design").
+        """
+        if body.content.entity_refs:
+            return True
+        return _WIRE_TO_DOMAIN_ZONE.get(body.content.zone_hint) is ZoneId.SEMANTIC
+
+    def _do_write_zone3_5(
+        caller: AuthenticatedCaller, body: sc.WriteMemoryRequest, source_type: SourceType
+    ) -> tuple[Response, str | None]:
+        """FR-013 design Section 4.2 steps 1-6: Zone 5 / Zone 3 routing and writes.
+
+        Cite FR-013 verbatim (SRS.md, FR-013 row): "On every write
+        targeting Zone 3 (Semantic) or Zone 5 (Entity), the system SHALL
+        check the candidate against existing records for the same
+        (entity, predicate) or (subject, predicate, object). On
+        contradiction, neither record SHALL be overwritten; both SHALL be
+        marked conflict_status=disputed and both confidences reduced,
+        until resolved by re-confirmation, explicit user correction, or
+        majority-source consensus."
+        """
+        retrieval_context_hash = body.provenance.retrieval_context_hash
+        if not retrieval_context_hash or not retrieval_context_hash.strip():
+            return _error(
+                400, "INVALID_REQUEST",
+                "provenance.retrieval_context_hash is required for a Zone 3/5 write (FR-013)",
+            ), None
+
+        if not ctx.postgres_available or ctx.semantic_repository is None or ctx.conflict_aware_entity_repository is None:
+            logger.error(
+                "Zone 3/5 write attempted with no live Postgres backend",
+                extra={"tenant_id": caller.tenant_id},
+            )
+            return _error(
+                503, "WRITE_REJECTED_NOT_DURABLE",
+                "zone '3-semantic'/'5-entity' is temporarily unavailable",
+            ), None
+
+        subjects = tuple(ref.entity_id for ref in body.content.entity_refs if ref.role == "subject")
+        objects = tuple(ref.entity_id for ref in body.content.entity_refs if ref.role == "object")
+
+        zone5_written = False
+        zone3_edges_written: list[str] = []
+        zone3_edges_failed: list[str] = []
+
+        try:
+            candidate_fact = CandidateFact(
+                tenant_id=caller.tenant_id,
+                subjects=subjects,
+                objects=objects,
+                predicate=body.content.predicate or "",
+                statement=body.content.text or "",
+                subject_scope=body.content.subject_scope or "",
+                index_reverse=body.content.index_reverse,
+            )
+
+            if len(subjects) == 1:
+                ctx.conflict_aware_entity_repository.write_attribute(
+                    caller.tenant_id,
+                    subjects[0],
+                    candidate_fact.predicate,
+                    candidate_fact.statement,
+                    str(uuid.uuid4()),
+                    source_type=source_type,
+                    retrieval_context_hash=retrieval_context_hash,
+                )
+                zone5_written = True
+
+            routing = classify(
+                candidate_fact,
+                edge_id_factory=lambda: str(uuid.uuid4()),
+                fact_id_factory=lambda: str(uuid.uuid4()),
+            )
+        except ValueError as exc:
+            return _error(400, "INVALID_REQUEST", str(exc)), None
+        except ZoneRepositoryError:
+            logger.error(
+                "Zone 5 (Entity) write failed",
+                extra={"tenant_id": caller.tenant_id},
+                exc_info=True,
+            )
+            return _error(
+                503, "WRITE_REJECTED_NOT_DURABLE",
+                "zone '5-entity' is temporarily unavailable",
+            ), None
+
+        if isinstance(routing, SemanticEdgeRouting):
+            for edge in routing.edges:
+                item_id = _zone3_edge_item_id(edge.subject_ref, edge.predicate, edge.object_ref)
+                try:
+                    ctx.semantic_repository.insert_edge(
+                        edge,
+                        provenance_id=str(uuid.uuid4()),
+                        source_type=source_type,
+                        retrieval_context_hash=retrieval_context_hash,
+                    )
+                    zone3_edges_written.append(item_id)
+                except ZoneRepositoryError:
+                    logger.error(
+                        "Zone 3 (Semantic) edge write failed",
+                        extra={"tenant_id": caller.tenant_id, "item_id": item_id},
+                        exc_info=True,
+                    )
+                    zone3_edges_failed.append(item_id)
+        elif isinstance(routing, GeneralFactRouting):
+            try:
+                ctx.semantic_repository.insert_general_fact(
+                    routing.fact,
+                    provenance_id=str(uuid.uuid4()),
+                    source_type=source_type,
+                    retrieval_context_hash=retrieval_context_hash,
+                )
+            except ZoneRepositoryError:
+                logger.error(
+                    "Zone 3 (Semantic) general fact write failed",
+                    extra={"tenant_id": caller.tenant_id},
+                    exc_info=True,
+                )
+                return _error(
+                    503, "WRITE_REJECTED_NOT_DURABLE",
+                    "zone '3-semantic' is temporarily unavailable",
+                ), None
+
+        write_id = str(uuid.uuid4())
+        status: Literal["accepted", "partial"] = "partial" if zone3_edges_failed else "accepted"
+        receipt = sc.WriteReceipt(
+            write_id=write_id,
+            status=status,
+            accepted_at=ctx.clock.now(),
+            zone5_written=zone5_written,
+            zone3_edges_written=zone3_edges_written,
+            zone3_edges_failed=zone3_edges_failed,
+        )
+        write_status_store.record(st.WriteStatusEntry(write_id=write_id, status="accepted", item_id=None))
+        status_code = 207 if zone3_edges_failed else 202
+        return JSONResponse(status_code=status_code, content=receipt.model_dump(mode="json")), write_id
+
     def _do_write(caller: AuthenticatedCaller, body: sc.WriteMemoryRequest) -> tuple[Response, str | None]:
-        if _resolve_source_type(body) is None:
+        resolved_source_type_raw = _resolve_source_type(body)
+        if resolved_source_type_raw is None:
             return _error(
                 422, "INVALID_SOURCE_TYPE",
                 "writeMemory requires a resolvable provenance.source_type (FR-010)",
             ), None
 
-        if body.content.entity_refs:
-            return _error(
-                422, "ZONE_3_5_PREDICATE_UNRESOLVABLE",
-                "openapi.yaml WriteMemoryRequest.content carries no 'predicate' field; "
-                "this host cannot construct a real SemanticEdge/GeneralFact for an "
-                "entity_refs-bearing write without fabricating one (flagged gap, "
-                "see api.app module docstring)",
-            ), None
+        if _zone3_5_routing_gate_triggered(body):
+            return _do_write_zone3_5(caller, body, SourceType(resolved_source_type_raw))
 
         write_id = str(uuid.uuid4())
         text = body.content.text or ""

src/dashanan/api/schemas.py

@@ -39,6 +39,7 @@ class ProvenanceWriteBlock(BaseModel):
     importance: float | None = None
     purpose: str
     subject_id: str | None = None
+    retrieval_context_hash: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
 
 
 class EntityRef(BaseModel):
@@ -55,6 +56,8 @@ class WriteMemoryContent(BaseModel):
     task_signature_hash: str | None = None
     entity_refs: list[EntityRef] = Field(default_factory=list)
     index_reverse: bool = False
+    predicate: str | None = None
+    subject_scope: str | None = None
 
 
 class WriteMemoryRequest(BaseModel):
@@ -65,8 +68,11 @@ class WriteMemoryRequest(BaseModel):
 
 class WriteReceipt(BaseModel):
     write_id: str
-    status: Literal["accepted"] = "accepted"
+    status: Literal["accepted", "partial"] = "accepted"
     accepted_at: datetime
+    zone5_written: bool = False
+    zone3_edges_written: list[str] = Field(default_factory=list)
+    zone3_edges_failed: list[str] = Field(default_factory=list)
 
 
 class BatchWriteReceiptItem(BaseModel):