From fceb5f27f09015d89e2723ccaf8f1879ca3a3480 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 4 Oct 2026 12:54:57 +0200 Subject: [PATCH] Receive source-owned depends_on edges through Forge projection (STATE-WP-0094-T05) Derive frontmatter and task-block depends_on edges from the pinned commit and reconcile workplan_dependencies under an explicit ownership contract: only forge: rows are created or removed, external relations are preserved, missing/ambiguous/self targets are recorded and hold owned edges, failures roll back, and repeat receipts are noop. Additive migration b3c4d5e6f7a8. Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 671493@bnt-lap001 Assistant-Session: 60d7b11f-78f9-4731-9367-918511b2762d --- api/models/workplan_dependency.py | 7 + api/schemas/workplan_dependency.py | 2 + api/services/forge_projection.py | 301 ++++++++++++++++++ docs/forge-dependency-edge-receiving.md | 84 +++++ ...e6f7a8_workplan_dependency_source_owner.py | 39 +++ tests/test_forge_dependency_edges.py | 204 ++++++++++++ 6 files changed, 637 insertions(+) create mode 100644 docs/forge-dependency-edge-receiving.md create mode 100644 migrations/versions/b3c4d5e6f7a8_workplan_dependency_source_owner.py create mode 100644 tests/test_forge_dependency_edges.py diff --git a/api/models/workplan_dependency.py b/api/models/workplan_dependency.py index 2836bcf..fdc5a89 100644 --- a/api/models/workplan_dependency.py +++ b/api/models/workplan_dependency.py @@ -91,6 +91,13 @@ class WorkplanDependency(Base, TimestampMixin): String(40), nullable=False, default="blocks", server_default="blocks", index=True ) description: Mapped[str | None] = mapped_column(Text, nullable=True) + # Source-edge ownership (STATE-WP-0094-T05, docs/forge-dependency-edge-receiving.md). + # ``forge:`` when normal Forge receiving derived this edge from + # that repository's files; NULL for externally authored relations (API, + # MCP, consistency fix path), which receiving never removes. + source_owner: Mapped[str | None] = mapped_column(String(160), nullable=True, index=True) + # Immutable source commit that first introduced the source-owned edge. + source_commit: Mapped[str | None] = mapped_column(String(64), nullable=True) from_workplan: Mapped["Workplan"] = relationship( # noqa: F821 "Workplan", foreign_keys=[from_workplan_id] diff --git a/api/schemas/workplan_dependency.py b/api/schemas/workplan_dependency.py index 31ec765..6cf1cd5 100644 --- a/api/schemas/workplan_dependency.py +++ b/api/schemas/workplan_dependency.py @@ -24,6 +24,8 @@ class WorkplanDependencyRead(BaseModel): to_task_id: uuid.UUID | None = None relationship_type: str description: str | None = None + source_owner: str | None = None + source_commit: str | None = None created_at: datetime updated_at: datetime diff --git a/api/services/forge_projection.py b/api/services/forge_projection.py index 28dd491..9af8884 100644 --- a/api/services/forge_projection.py +++ b/api/services/forge_projection.py @@ -138,6 +138,10 @@ class DerivedTask: decision_id: str | None = None blocked_on: str | None = None wait_until: date | None = None + # Task-block ``depends_on`` targets (workplan or task record ids), in + # source order, deduplicated. Unqualified ``T##`` targets are qualified + # with the owning workplan. + depends_on: list[str] = field(default_factory=list) @dataclass @@ -155,6 +159,10 @@ class DerivedWorkplan: flavor_promotion_reason: str | None = None flavor_promoted_from: str | None = None historical_preservation: bool = False + # Frontmatter ``depends_on`` / ``depends_on_workplans`` / ``depends_on_tasks`` + # split as C-20 splits them: task record ids vs workplan record ids. + depends_on_workplans: list[str] = field(default_factory=list) + depends_on_tasks: list[str] = field(default_factory=list) @dataclass @@ -207,6 +215,8 @@ class DerivedProjection: "relative_path": w.relative_path, "archived": w.archived, "historical_preservation": w.historical_preservation, + "depends_on_workplans": list(w.depends_on_workplans), + "depends_on_tasks": list(w.depends_on_tasks), "tasks": [ { "record_id": t.record_id, @@ -222,6 +232,7 @@ class DerivedProjection: "decision_id": t.decision_id, "blocked_on": t.blocked_on, "wait_until": t.wait_until.isoformat() if t.wait_until else None, + "depends_on": list(t.depends_on), } for t in w.tasks ], @@ -353,6 +364,50 @@ def _source_needs_human(value: Any) -> bool: raise ForgeDeriveError("Source needs_human must be boolean; no projection was applied") +_TASK_TARGET_RE = re.compile(r"-T\d{2,3}$") +_SHORT_TASK_TARGET_RE = re.compile(r"^T\d{2,3}$", re.IGNORECASE) + + +def _source_dependency_list(value: Any, field_name: str) -> list[str]: + """Return declared dependency targets, refusing shapes C-20 cannot read. + + A mapping or nested list is not a dependency declaration; dropping it would + silently lose an edge, so the whole derive is refused instead. + """ + if value is None or value == "": + return [] + if isinstance(value, str): + items = value.split(",") + elif isinstance(value, list): + if any(isinstance(item, (dict, list)) for item in value): + raise ForgeDeriveError(f"Malformed source {field_name}; no projection was applied") + items = [str(item) for item in value if item is not None] + elif isinstance(value, dict): + raise ForgeDeriveError(f"Malformed source {field_name}; no projection was applied") + else: + items = [str(value)] + out: list[str] = [] + for item in items: + target = item.strip().strip('"').strip("'").strip() + if target and target not in out: + out.append(target) + return out + + +def _split_dependency_targets(meta: dict, owner_record_id: str) -> tuple[list[str], list[str]]: + """Split declared targets into (workplan ids, task ids) like C-20 does.""" + declared = _source_dependency_list(meta.get("depends_on"), "depends_on") + workplans = [t for t in _source_dependency_list(meta.get("depends_on_workplans"), "depends_on_workplans")] + tasks = _source_dependency_list(meta.get("depends_on_tasks"), "depends_on_tasks") + for target in declared: + if _SHORT_TASK_TARGET_RE.match(target): + target = qualify_task_id(target, owner_record_id) or target + (tasks if _TASK_TARGET_RE.search(target) else workplans).append(target) + tasks = [qualify_task_id(t, owner_record_id) or t if _SHORT_TASK_TARGET_RE.match(t) else t for t in tasks] + dedupe = lambda items: list(dict.fromkeys(items)) # noqa: E731 + return dedupe(workplans), dedupe(tasks) + + def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]: headings = [ (m.start(), len(m.group(1)), m.group(2).strip()) for m in _HEADING_RE.finditer(body) @@ -421,11 +476,17 @@ def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]: decision_id=_source_wait_text(block.get("decision_id")), blocked_on=_source_wait_text(block.get("blocked_on")), wait_until=_source_wait_until(block.get("wait_until")), + depends_on=_task_dependency_targets(block, workplan_id), ) ) return out +def _task_dependency_targets(block: dict, workplan_id: str) -> list[str]: + workplans, tasks = _split_dependency_targets(block, workplan_id) + return list(dict.fromkeys(workplans + tasks)) + + def _workplan_description(body: str) -> str | None: """Return bounded prose under ``## Goal`` when one is present.""" match = re.search(r"^##\s+Goal\s*$", body, re.MULTILINE | re.IGNORECASE) @@ -492,6 +553,7 @@ def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> Derive rid = str(meta.get("id") or "").strip() if not rid: continue + frontmatter_deps = _split_dependency_targets(meta, rid) proj.workplans.append( DerivedWorkplan( record_id=rid, @@ -510,6 +572,8 @@ def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> Derive flavor=_source_flavor(meta.get("flavor")), flavor_promotion_reason=_source_promotion_reason(meta.get("flavor_promotion_reason")), flavor_promoted_from=_source_flavor(meta.get("flavor_promoted_from"), "flavor_promoted_from"), + depends_on_workplans=frontmatter_deps[0], + depends_on_tasks=frontmatter_deps[1], ) ) proj.workplans.sort(key=lambda w: w.record_id) @@ -854,6 +918,15 @@ class ResetOutcome: created_tasks: list[str] = field(default_factory=list) updated_tasks: list[str] = field(default_factory=list) cancelled_tasks: list[str] = field(default_factory=list) + # Source-owned dependency edges (STATE-WP-0094-T05). + created_edges: list[dict[str, Any]] = field(default_factory=list) + removed_edges: list[dict[str, Any]] = field(default_factory=list) + # Declared edges whose target is missing, ambiguous or invalid: recorded, + # never silently dropped; the from-side's owned edges are retained. + unresolved_edges: list[dict[str, Any]] = field(default_factory=list) + # Declared edges already present as externally authored rows: preserved + # and not adopted, so receiving can never remove them. + external_edges: list[dict[str, Any]] = field(default_factory=list) notes: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: @@ -871,6 +944,10 @@ class ResetOutcome: "created_tasks": len(self.created_tasks), "updated_tasks": len(self.updated_tasks), "cancelled_tasks": len(self.cancelled_tasks), + "created_edges": len(self.created_edges), + "removed_edges": len(self.removed_edges), + "unresolved_edges": len(self.unresolved_edges), + "external_edges": len(self.external_edges), }, "created": self.created, "updated": self.updated, @@ -879,6 +956,10 @@ class ResetOutcome: "created_tasks": self.created_tasks, "updated_tasks": self.updated_tasks, "cancelled_tasks": self.cancelled_tasks, + "created_edges": self.created_edges, + "removed_edges": self.removed_edges, + "unresolved_edges": self.unresolved_edges, + "external_edges": self.external_edges, "refused": self.refused, "notes": self.notes, } @@ -1483,8 +1564,11 @@ async def reset_repository_projection( ) return outcome + received_rows: dict[str, Any] = {} for key, w in want.items(): row = matched.get(key) + if row is not None: + received_rows[key] = row if row is None: row = Workplan( id=uuid.UUID(w.uuid), @@ -1505,6 +1589,7 @@ async def reset_repository_projection( ) session.add(row) await session.flush() + received_rows[key] = row for t in w.tasks: # Safe only because nothing exists to mis-match against: this # workplan is new to the hub. @@ -1584,6 +1669,8 @@ async def reset_repository_projection( r.projection_retired_reason = RETIRE_REASON r.slug = _tombstone_slug(r.slug or str(r.id), now) + await _reconcile_source_edges(session, repo, repo_slug, derived, want, received_rows, outcome) + if ( outcome.created or outcome.updated @@ -1592,11 +1679,225 @@ async def reset_repository_projection( or outcome.created_tasks or outcome.updated_tasks or outcome.cancelled_tasks + or outcome.created_edges + or outcome.removed_edges ): outcome.status = "applied" return outcome +# --------------------------------------------------------------------------- +# Source-owned dependency edges (STATE-WP-0094-T05) +# docs/forge-dependency-edge-receiving.md +# --------------------------------------------------------------------------- + +EDGE_SOURCE_PREFIX = "forge:" +_CLOSED_TASK_STATUSES = {"done", "cancel"} + + +def _edge_workplan_closed(status: str | None) -> bool: + from api.workplan_status import CLOSED_WORKPLAN_STATUSES, normalize_workplan_status + + try: + return normalize_workplan_status(status or "") in CLOSED_WORKPLAN_STATUSES + except ValueError: + return False + + +def _edge_task_closed(status: str | None) -> bool: + from api.task_status import normalize_task_status + + try: + return normalize_task_status(status or "todo", default="todo") in _CLOSED_TASK_STATUSES + except ValueError: + return False + + +def edge_source_owner(repo_slug: str) -> str: + return f"{EDGE_SOURCE_PREFIX}{repo_slug}" + + +async def _resolve_edge_target(session: Any, record_id: str, cache: dict) -> tuple[str, Any]: + """Resolve a declared target fleet-wide to ``("ok", uuid)`` or a reason. + + Workplans match by ADR-007 derived UUID or record-id slug; tasks by derived + UUID or ``record_id``. Projection-retired workplans (and their tasks) are + not targets. More than one candidate is ambiguous: no winner is chosen. + """ + if record_id in cache: + return cache[record_id] + from sqlalchemy import func, or_, select + + from api.models.task import Task + from api.models.workplan import Workplan + + derived = uuid.UUID(derived_record_uuid(record_id)) + if _TASK_TARGET_RE.search(record_id): + stmt = ( + select(Task.id) + .join(Workplan, Workplan.id == Task.workplan_id) + .where(or_(Task.id == derived, Task.record_id == record_id)) + .where(Workplan.projection_retired_at.is_(None)) + ) + else: + stmt = select(Workplan.id).where( + or_(Workplan.id == derived, func.lower(Workplan.slug) == record_id.lower()) + ).where(Workplan.projection_retired_at.is_(None)) + ids = sorted({row[0] for row in (await session.execute(stmt)).all()}, key=str) + if not ids: + result: tuple[str, Any] = ("missing", None) + elif len(ids) > 1: + result = ("ambiguous", [str(i) for i in ids]) + else: + result = ("ok", ids[0]) + cache[record_id] = result + return result + + +async def _reconcile_source_edges( + session: Any, + repo: Any, + repo_slug: str, + derived: DerivedProjection, + want: dict[str, DerivedWorkplan], + received_rows: dict[str, Any], + outcome: ResetOutcome, +) -> None: + """Reconcile ``workplan_dependencies`` rows owned by this repository's source. + + Ownership: a row is source-owned iff ``source_owner == forge:`` + and its from-side belongs to this repository. Only those rows are created or + removed here. Rows with any other (or no) owner are externally authored and + preserved, even when they duplicate a declared edge. + """ + from sqlalchemy import select + + from api.models.task import Task + from api.models.workplan import Workplan + from api.models.workplan_dependency import WorkplanDependency + + owner = edge_source_owner(repo_slug) + await session.flush() + + # (from_workplan_id, from_task_id, to_workplan_id, to_task_id, relationship) + desired: dict[tuple, dict[str, Any]] = {} + # From-sides with an unresolved declaration keep every owned edge: the + # unresolved target may be one of them (e.g. newly ambiguous). + held_sides: set[tuple] = set() + cache: dict[str, tuple[str, Any]] = {} + + async def _declare(row: Any, w: DerivedWorkplan, from_task: Any, from_label: str, target: str) -> None: + side = (row.id, from_task.id if from_task is not None else None) + is_task = bool(_TASK_TARGET_RE.search(target)) + state, value = await _resolve_edge_target(session, target, cache) + base = {"workplan": w.record_id, "from": from_label, "target": target} + if state == "ok" and ((not is_task and value == row.id) or (is_task and from_task is not None and value == from_task.id)): + state, value = "self", None + if state != "ok": + held_sides.add(side) + outcome.unresolved_edges.append({**base, "reason": state, **({"candidates": value} if value else {})}) + return + rel = "starts_after" if is_task else "blocks" + key = (row.id, side[1], None if is_task else value, value if is_task else None, rel) + desired.setdefault(key, {**base, "relationship_type": rel}) + + for key, w in want.items(): + row = received_rows.get(key) + if row is None or w.historical_preservation: + continue + for target in w.depends_on_workplans + w.depends_on_tasks: + await _declare(row, w, None, w.record_id, target) + # C-20 parity: closed workplans and closed tasks are history, not live + # ordering, so their task-block edges are not desired. + if _edge_workplan_closed(w.status): + continue + task_rows = None + for t in w.tasks: + if not t.depends_on or _edge_task_closed(t.status): + continue + if task_rows is None: + task_rows = { + (tr.record_id or "").strip().lower(): tr + for tr in (await session.execute(select(Task).where(Task.workplan_id == row.id))).scalars() + } + from_task = task_rows.get(t.record_id.strip().lower()) + if from_task is None: + held_sides.add((row.id, "*")) + for target in t.depends_on: + outcome.unresolved_edges.append( + {"workplan": w.record_id, "from": t.record_id, "target": target, + "reason": "from-task-missing"} + ) + continue + for target in t.depends_on: + await _declare(row, w, from_task, t.record_id, target) + + repo_workplan_ids = list( + (await session.execute(select(Workplan.id).where(Workplan.repo_id == repo.id))).scalars() + ) + existing = list( + ( + await session.execute( + select(WorkplanDependency).where( + WorkplanDependency.from_workplan_id.in_(repo_workplan_ids) + ) + ) + ).scalars() + ) if repo_workplan_ids else [] + by_key: dict[tuple, Any] = { + (e.from_workplan_id, e.from_task_id, e.to_workplan_id, e.to_task_id, e.relationship_type): e + for e in existing + } + + def _edge_dict(edge_key: tuple, info: dict[str, Any] | None = None) -> dict[str, Any]: + from_wp, from_task, to_wp, to_task, rel = edge_key + return { + **(info or {}), + "from_workplan_id": str(from_wp), + "from_task_id": str(from_task) if from_task else None, + "to_workplan_id": str(to_wp) if to_wp else None, + "to_task_id": str(to_task) if to_task else None, + "relationship_type": rel, + } + + for edge_key in sorted(desired, key=str): + info = desired[edge_key] + current = by_key.get(edge_key) + if current is not None: + if current.source_owner != owner: + outcome.external_edges.append(_edge_dict(edge_key, info)) + continue + from_wp, from_task, to_wp, to_task, rel = edge_key + session.add( + WorkplanDependency( + from_workplan_id=from_wp, + from_task_id=from_task, + to_workplan_id=to_wp, + to_task_id=to_task, + relationship_type=rel, + description=f"{info['from']} depends_on {info['target']}", + source_owner=owner, + source_commit=derived.commit, + ) + ) + outcome.created_edges.append(_edge_dict(edge_key, info)) + + for edge_key, edge in sorted(by_key.items(), key=lambda kv: str(kv[0])): + if edge.source_owner != owner or edge_key in desired: + continue + if (edge.from_workplan_id, edge.from_task_id) in held_sides or ( + edge.from_workplan_id, "*" + ) in held_sides: + outcome.notes.append( + f"Retained source-owned edge {edge.id}: its from-side has an unresolved declaration." + ) + continue + await session.delete(edge) + outcome.removed_edges.append(_edge_dict(edge_key, {"id": str(edge.id)})) + # Integrity failures surface here, inside the caller's transaction. + await session.flush() + + # --------------------------------------------------------------------------- # Fleet form (STATE-WP-0083-T04) # --------------------------------------------------------------------------- diff --git a/docs/forge-dependency-edge-receiving.md b/docs/forge-dependency-edge-receiving.md new file mode 100644 index 0000000..b3ec5f2 --- /dev/null +++ b/docs/forge-dependency-edge-receiving.md @@ -0,0 +1,84 @@ +# Forge dependency-edge receiving (STATE-WP-0094-T05) + +Status: source implementation landed 2026-10-04; not deployed. Deployment and +any receiving schedule need separate founder authorization. + +## Problem + +`docs/owner-return-handoff.md` ("T05 graph freshness assessment") records that +normal Forge receiving (`api/services/forge_projection.py`) did not derive +`depends_on` edges or reconcile `workplan_dependencies`. Only C-20 created +edges, from a workstation checkout, and nothing ever removed an edge whose +declaration disappeared. Adding a projection that "replaces every dependency +row" would delete externally authored relations, so receiving needs an explicit +ownership contract first. + +## Contract + +1. **Source.** Edges are derived only from the immutable commit that receiving + already pins (`expected_commit` = derived commit), from the same files as the + rest of the projection: + - workplan frontmatter `depends_on`, `depends_on_workplans`, + `depends_on_tasks` - split like C-20: ids ending `-T\d{2,3}` are task + targets (`starts_after`), all others workplan targets (`blocks`); + - task-block `depends_on` - the waiting task is the from side + (`from_task_id`). As in C-20, task-block edges of closed workplans + (`finished`/`archived`) and closed tasks (`done`/`cancel`) are history and + not desired. + - A bare `T##` target is qualified with the declaring workplan id. + - Bound historical workplans (`historical_preservation`) declare no edges. + - A declaration that is a mapping or nested list refuses the whole derive + (`ForgeDeriveError`); it is never dropped. +2. **Ownership.** A row is source-owned iff + `workplan_dependencies.source_owner = 'forge:'` (additive, + nullable column, migration `b3c4d5e6f7a8`) and its from-workplan belongs to + that repository. `source_commit` records the commit that introduced it. + Receiving creates rows only with that owner and removes only rows with that + owner. Rows with NULL or any other owner (API, MCP, C-20 fix path, other + tools) are externally authored and are never modified or removed. +3. **Already present externally.** When a declared edge already exists as an + external row (same from workplan, from task, target and relationship), it is + preserved, not adopted, and reported in `external_edges`. Adoption is a + deliberate future decision (see risks). +4. **Target resolution (cross-repository).** Fleet-wide, excluding + projection-retired workplans and their tasks: workplans by ADR-007 derived + UUID or record-id slug; tasks by derived UUID or `record_id`. Exactly one + candidate resolves. Targets created in the same receipt resolve, because + resolution runs after the workplan/task writes are flushed. +5. **Missing / ambiguous / invalid targets.** Zero candidates = `missing`, more + than one = `ambiguous` (candidates listed; no winner chosen), a self edge = + `self`, a waiting task the hub lacks = `from-task-missing`. Each is recorded + in `unresolved_edges`; nothing is silently dropped. The receipt itself still + applies (a target in a not-yet-received repository must not block the + source repository's workplan projection). +6. **Removal.** Only source-owned rows whose key is no longer declared are + removed - including rows of workplans retired in the same acknowledged + receipt. If the from side (workplan or waiting task) has any unresolved + declaration, its owned rows are retained and a note is emitted, because the + unresolved target may be one of them. +7. **Transactional failure.** Edge reconciliation runs inside the same session + as the workplan/task projection and flushes at the end. Any error propagates + and the endpoint commits nothing; a refused receipt returns before edges are + touched. +8. **Repeat noop.** Owned rows are not rewritten when unchanged, so the same + commit received twice reports `status: noop` with zero created/removed edges. + (`unresolved_edges` is still reported every time.) + +Outcome schema additions (`state-hub.projection-reset.v1`): `created_edges`, +`removed_edges`, `unresolved_edges`, `external_edges` lists and their counts. + +## Coexistence with C-20 + +C-20 keys existence on (from task, target, relationship) regardless of owner, +so it does not duplicate a source-owned edge; edges its fix path creates are +external to receiving. Existing hub edges therefore stay external until an +explicit adoption decision. + +## Open risks + +- Pre-existing C-20/API edges are never adopted, so a removed declaration does + not remove an edge C-20 created earlier; adoption needs its own decision. +- `diff_against_hub` (dry run) does not yet preview edge changes. +- Ambiguity by slug for workplans is mostly prevented by the global slug index; + task `record_id` ambiguity across workplans is the realistic case. +- Not deployed; no evidence of scheduled freshness. T05 stays in progress. diff --git a/migrations/versions/b3c4d5e6f7a8_workplan_dependency_source_owner.py b/migrations/versions/b3c4d5e6f7a8_workplan_dependency_source_owner.py new file mode 100644 index 0000000..9464765 --- /dev/null +++ b/migrations/versions/b3c4d5e6f7a8_workplan_dependency_source_owner.py @@ -0,0 +1,39 @@ +"""workplan dependency source ownership (STATE-WP-0094-T05) + +Additive: ``workplan_dependencies.source_owner`` marks edges that normal Forge +receiving derived from a repository's files (``forge:``) and +``source_commit`` records the immutable commit that introduced them. Existing +rows stay NULL, i.e. externally authored, and are never removed by receiving. + +Revision ID: b3c4d5e6f7a8 +Revises: a0b1c2d3e4f5 +""" +import sqlalchemy as sa +from alembic import op + +revision = "b3c4d5e6f7a8" +down_revision = "a0b1c2d3e4f5" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "workplan_dependencies", + sa.Column("source_owner", sa.String(length=160), nullable=True), + ) + op.add_column( + "workplan_dependencies", + sa.Column("source_commit", sa.String(length=64), nullable=True), + ) + op.create_index( + "ix_workplan_dependencies_source_owner", + "workplan_dependencies", + ["source_owner"], + ) + + +def downgrade() -> None: + op.drop_index("ix_workplan_dependencies_source_owner", table_name="workplan_dependencies") + op.drop_column("workplan_dependencies", "source_commit") + op.drop_column("workplan_dependencies", "source_owner") diff --git a/tests/test_forge_dependency_edges.py b/tests/test_forge_dependency_edges.py new file mode 100644 index 0000000..f45d86c --- /dev/null +++ b/tests/test_forge_dependency_edges.py @@ -0,0 +1,204 @@ +"""Source-owned dependency edges through normal Forge receiving (STATE-WP-0094-T05). + +Runs the real reconcile endpoint against the PostgreSQL test database, as the +qualifier receiving tests do. Contract: docs/forge-dependency-edge-receiving.md. +""" +from pathlib import Path + +import pytest + +from api.services import forge_projection as fp +from tests.conftest import create_test_domain, create_test_repo, create_test_workplan + +TARGET = """--- +id: TGT-WP-0001 +type: workplan +title: Target +status: active +--- + +## Target task + +```task +id: TGT-WP-0001-T01 +status: todo +``` +""" + + +def _source(frontmatter_deps: str, wp2_deps: str = "", t02_target: str = "MISSING-WP-0009") -> str: + return { + "workplans/SRC-WP-0001.md": ( + "---\nid: SRC-WP-0001\ntype: workplan\ntitle: Source\nstatus: active\n" + + frontmatter_deps + "---\n\n## Wait\n\n```task\nid: SRC-WP-0001-T01\nstatus: wait\n" + "depends_on: [SRC-WP-0002]\n```\n\n## Missing\n\n```task\nid: T02\nstatus: wait\n" + f"depends_on: {t02_target}\n```\n\n## Closed\n\n```task\nid: SRC-WP-0001-T03\n" + "status: done\ndepends_on: TGT-WP-0001\n```\n" + ), + "workplans/SRC-WP-0002.md": ( + "---\nid: SRC-WP-0002\ntype: workplan\ntitle: Sibling\nstatus: active\n" + + wp2_deps + "---\n" + ), + } + + +def _write(root: Path, files: dict[str, str]) -> None: + for rel, text in files.items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + + +@pytest.mark.asyncio +async def test_receiving_owns_only_source_edges(client, monkeypatch, tmp_path): + await create_test_domain(client) + await create_test_repo(client, slug="edge-target") + await create_test_repo(client, slug="edge-source") + from api.config import settings + from api.routers import work_record_projection as route + + monkeypatch.setattr(settings, "state_hub_instance_role", "primary") + monkeypatch.setattr(settings, "state_hub_instance_label", "railiance01") + + roots = {"edge-target": tmp_path / "target", "edge-source": tmp_path / "source"} + revision = {"edge-target": "a" * 40, "edge-source": "b" * 40} + _write(roots["edge-target"], {"workplans/TGT-WP-0001.md": TARGET}) + _write(roots["edge-source"], _source("depends_on: [TGT-WP-0001, TGT-WP-0001-T01]\n")) + monkeypatch.setattr( + route, "derive_from_forge", + lambda slug: fp.derive_from_checkout(roots[slug], slug, revision[slug]), + ) + + async def reconcile(slug, key): + response = await client.post( + f"/repos/{slug}/work-record-projection/reconcile", + json={"expected_commit": revision[slug]}, headers={"Idempotency-Key": key}, + ) + assert response.status_code == 200, response.text + return response.json()["outcome"] + + async def edges(record_id): + wp_id = fp.derived_record_uuid(record_id) + response = await client.get(f"/workplans/{wp_id}/dependencies/") + assert response.status_code == 200, response.text + return [row for row in response.json() if row["from_workplan_id"] == wp_id] + + def counts(outcome): + c = outcome["counts"] + return (c["created_edges"], c["removed_edges"], c["unresolved_edges"], c["external_edges"]) + + await reconcile("edge-target", "edges:target") + first = await reconcile("edge-source", "edges:first") + assert first["status"] == "applied" + # wp->wp, wp->task, task->sibling wp; T02 target missing; closed T03 not desired. + assert counts(first) == (3, 0, 1, 0) + assert first["unresolved_edges"][0] == { + "workplan": "SRC-WP-0001", "from": "SRC-WP-0001-T02", + "target": "MISSING-WP-0009", "reason": "missing", + } + rows = await edges("SRC-WP-0001") + assert len(rows) == 3 + assert {r["source_owner"] for r in rows} == {"forge:edge-source"} + assert {r["source_commit"] for r in rows} == {"b" * 40} + assert {r["relationship_type"] for r in rows} == {"blocks", "starts_after"} + assert sum(1 for r in rows if r["from_task_id"]) == 1 + + # Repeat of the same immutable commit is a noop. + repeat = await reconcile("edge-source", "edges:repeat") + assert repeat["status"] == "noop" + assert counts(repeat) == (0, 0, 1, 0) + + # Externally authored relations: one undeclared, one the source will declare. + sibling = fp.derived_record_uuid("SRC-WP-0002") + target_wp = fp.derived_record_uuid("TGT-WP-0001") + other_repo = await create_test_repo(client, slug="edge-other") + other = await create_test_workplan(client, other_repo["id"], slug="other-wp", title="Other") + for to_id in (target_wp, other["id"]): + r = await client.post(f"/workplans/{sibling}/dependencies/", + json={"to_workplan_id": to_id, "relationship_type": "blocks"}) + assert r.status_code == 201, r.text + + # Source drops the task-target edge and declares the externally present one. + _write(roots["edge-source"], _source( + "depends_on: [TGT-WP-0001]\n", wp2_deps="depends_on: TGT-WP-0001\n")) + revision["edge-source"] = "c" * 40 + second = await reconcile("edge-source", "edges:second") + assert second["status"] == "applied" + assert counts(second) == (0, 1, 1, 1) + assert second["removed_edges"][0]["to_task_id"] is not None + assert len(await edges("SRC-WP-0001")) == 2 + sibling_edges = await edges("SRC-WP-0002") + assert len(sibling_edges) == 2 and all(r["source_owner"] is None for r in sibling_edges) + assert (await reconcile("edge-source", "edges:second-repeat"))["status"] == "noop" + + # Dropping the declaration of the external edge does not remove it. + _write(roots["edge-source"], _source("depends_on: [TGT-WP-0001]\n")) + revision["edge-source"] = "d" * 40 + third = await reconcile("edge-source", "edges:third") + assert counts(third) == (0, 0, 1, 0) + assert len(await edges("SRC-WP-0002")) == 2 + + # Ambiguous target: refuse to choose, and retain the from-side's owned edges. + r = await client.post("/tasks/", json={"workplan_id": other["id"], "title": "Clash", + "record_id": "TGT-WP-0001-T01"}) + assert r.status_code == 201, r.text + clash_task = r.json()["id"] + target_task = fp.derived_record_uuid("TGT-WP-0001-T01") + _write(roots["edge-source"], _source("depends_on: [TGT-WP-0001-T01]\n", t02_target="SRC-WP-0002")) + revision["edge-source"] = "e" * 40 + fourth = await reconcile("edge-source", "edges:fourth") + assert counts(fourth) == (1, 0, 1, 0) + ambiguous = fourth["unresolved_edges"][0] + assert ambiguous["reason"] == "ambiguous" and ambiguous["target"] == "TGT-WP-0001-T01" + assert set(ambiguous["candidates"]) == {target_task, clash_task} + # The undeclared wp->TGT edge is held, not removed, while the side is unresolved. + assert len(await edges("SRC-WP-0001")) == 3 + + # Transactional failure: an error mid-reconcile leaves nothing applied. + files = _source("", t02_target="SRC-WP-0002") + files["workplans/SRC-WP-0001.md"] = files["workplans/SRC-WP-0001.md"].replace( + "title: Source\n", "title: Source renamed\n") + _write(roots["edge-source"], files) + revision["edge-source"] = "f" * 40 + calls = {"n": 0} + real_flush_target = fp._resolve_edge_target + + async def failing(session, record_id, cache): + calls["n"] += 1 + raise RuntimeError("injected edge resolution failure") + + monkeypatch.setattr(fp, "_resolve_edge_target", failing) + with pytest.raises(RuntimeError): + await client.post("/repos/edge-source/work-record-projection/reconcile", + json={"expected_commit": revision["edge-source"]}, + headers={"Idempotency-Key": "edges:fail"}) + assert calls["n"] >= 1 + monkeypatch.setattr(fp, "_resolve_edge_target", real_flush_target) + assert len(await edges("SRC-WP-0001")) == 3 + wp = (await client.get(f"/workplans/{fp.derived_record_uuid('SRC-WP-0001')}")).json() + assert wp["title"] == "Source" + + # The same commit then applies cleanly: frontmatter edge removed only. + final = await reconcile("edge-source", "edges:final") + assert counts(final) == (0, 1, 0, 0) + assert "SRC-WP-0001" in final["updated"] + assert len(await edges("SRC-WP-0001")) == 2 + assert (await reconcile("edge-source", "edges:final-repeat"))["status"] == "noop" + + +def test_malformed_dependency_declaration_refuses_derivation(tmp_path): + _write(tmp_path, {"workplans/SRC-WP-0001.md": ( + "---\nid: SRC-WP-0001\ntype: workplan\nstatus: active\ndepends_on: {a: b}\n---\n")}) + with pytest.raises(fp.ForgeDeriveError): + fp.derive_from_checkout(tmp_path, "edge-source", "a" * 40) + + +def test_dependency_targets_split_and_qualify(tmp_path): + _write(tmp_path, _source("depends_on: [TGT-WP-0001, TGT-WP-0001-T01, T04]\n" + "depends_on_workplans: TGT-WP-0001\n")) + derived = fp.derive_from_checkout(tmp_path, "edge-source", "a" * 40) + wp = {w.record_id: w for w in derived.workplans}["SRC-WP-0001"] + assert wp.depends_on_workplans == ["TGT-WP-0001"] + assert wp.depends_on_tasks == ["TGT-WP-0001-T01", "SRC-WP-0001-T04"] + assert [t.depends_on for t in wp.tasks] == [["SRC-WP-0002"], ["MISSING-WP-0009"], ["TGT-WP-0001"]] + assert derived.to_dict()["workplans"][0]["depends_on_tasks"] == ["TGT-WP-0001-T01", "SRC-WP-0001-T04"]