"""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"]