fix: reject non-zero delegation_depth on POST/PUT (parity with delega-api#28) (#22)

Self-hosted backend's task handlers were coherent by construction —
TaskCreate / TaskUpdate Pydantic schemas don't include delegation_depth,
parent_task_id, or root_task_id, so Pydantic v2 silently drops them.
POST /tasks always sets root_task_id atomically (either from parent or
self), and chain rows are only ever built via POST /tasks/:id/delegate.

But silent-drop is indistinguishable from "accepted and honored" from a
client's perspective. Hosted API (delega-api#28) landed explicit 400s
for the same shapes, making the contract visible. This change mirrors
that behavior so clients see identical 400s across deployments.

Added:
  - reject_chain_fields_on_create / _on_update: FastAPI Depends that
    inspect the raw request body and raise HTTPException(400) when:
      * POST /tasks has delegation_depth != 0 or any root_task_id
      * PUT /tasks/:id has any delegation_depth, parent_task_id, or
        root_task_id (chain fields are immutable once set)
    Error messages match hosted's wording, pointing callers to
    POST /tasks/:id/delegate for chain creation.
  - migrations/006_fix_orphan_chains.py: idempotent sqlite3 script
    that zeroes delegation_depth + re-opens status='delegated' for any
    orphan rows (parent/root NULL but depth > 0). No-op if the DB is
    already coherent, which is the expected case.
  - tests/test_task_coherence.py: 10 pytest cases covering bare create,
    non-zero depth reject, root_task_id reject on POST and PUT,
    parent_task_id still OK on POST but rejected on PUT, regular
    content updates, and /delegate atomicity.

Parent_task_id on POST /tasks is intentionally still accepted — that's
the documented path for chain creation via bare POST, and the handler
already sets parent + root + depth atomically in that branch.

Verified: pytest backend/tests/ passes 47/47 (10 new + 37 existing
permissions tests unchanged).

Co-authored-by: Marty (Clawdbot) <marty@mcmillan.io>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

backend/main.py

@@ -341,6 +341,52 @@ def get_task_for_agent(
     return task
 
 
+async def _reject_chain_fields(request: Request, *, mode: str) -> None:
+    # Chain membership (parent_task_id + root_task_id + delegation_depth) is
+    # set atomically by the server — via POST /tasks/:id/delegate for chain
+    # tasks, or implicitly on bare POST /tasks where root_task_id = self.id.
+    # Clients that set these fields directly can create orphan rows
+    # (delegation_depth > 0 with NULL parent/root), which this guard prevents.
+    # Parity with delega-api's coherence fix.
+    try:
+        raw = await request.json()
+    except Exception:
+        return  # Let FastAPI's own body parsing raise the user-facing error.
+    if not isinstance(raw, dict):
+        return
+    if "root_task_id" in raw:
+        raise HTTPException(
+            status_code=400,
+            detail="root_task_id cannot be set via this endpoint. Use POST /tasks/:id/delegate to create a chain task with parent_task_id and root_task_id set atomically.",
+        )
+    dd = raw.get("delegation_depth")
+    if mode == "create":
+        if dd is not None and dd != 0:
+            raise HTTPException(
+                status_code=400,
+                detail="delegation_depth cannot be set via POST /tasks. Use POST /tasks/:id/delegate to create a chain task with parent_task_id and root_task_id set atomically.",
+            )
+    elif mode == "update":
+        if "delegation_depth" in raw:
+            raise HTTPException(
+                status_code=400,
+                detail="delegation_depth is immutable once set. Chain membership is only established via POST /tasks/:id/delegate.",
+            )
+        if "parent_task_id" in raw:
+            raise HTTPException(
+                status_code=400,
+                detail="parent_task_id is immutable once set. Chain membership cannot be re-parented via PUT /tasks/:id.",
+            )
+
+
+async def reject_chain_fields_on_create(request: Request) -> None:
+    await _reject_chain_fields(request, mode="create")
+
+
+async def reject_chain_fields_on_update(request: Request) -> None:
+    await _reject_chain_fields(request, mode="update")
+
+
 @app.middleware("http")
 async def rate_limit_middleware(request: Request, call_next):
     path = request.url.path
@@ -1070,6 +1116,7 @@ def create_task(
     request: Request,
     db: Session = Depends(get_db),
     agent: Optional[models.Agent] = Depends(get_current_agent),
+    _coherence: None = Depends(reject_chain_fields_on_create),
 ):
     # Optional dedup check via header
     dedup_header = request.headers.get("X-Dedup-Check", "").lower()
@@ -1144,6 +1191,7 @@ def update_task(
     task: schemas.TaskUpdate,
     db: Session = Depends(get_db),
     agent: Optional[models.Agent] = Depends(get_current_agent),
+    _coherence: None = Depends(reject_chain_fields_on_update),
 ):
     db_task = get_task_for_agent(db, task_id, agent, require_mutation=True)
 

backend/migrations/006_fix_orphan_chains.py

@@ -0,0 +1,69 @@
+"""
+Migration 006: Fix orphan chain rows.
+
+Parity with the delega-api coherence fix (hosted: delega-dev/delega-api#28).
+Zeroes delegation_depth and re-opens status for any rows where
+delegation_depth > 0 but parent_task_id / root_task_id are NULL — those
+rows could not have been created via the server's /tasks or
+/tasks/:id/delegate handlers in their current form, but may exist from
+direct SQL writes, prior buggy API versions, or imports.
+
+Idempotent: reports "no orphans" and exits cleanly if the DB is already
+coherent, which is the expected case for installs minted after the
+handler-level guards landed in the same PR as this migration.
+
+Usage:
+    python backend/migrations/006_fix_orphan_chains.py
+    DELEGA_DB_PATH=/path/to/custom.db python backend/migrations/006_fix_orphan_chains.py
+"""
+import sqlite3
+import os
+
+
+def migrate(db_path: str):
+    conn = sqlite3.connect(db_path)
+    cur = conn.cursor()
+
+    cur.execute(
+        """
+        SELECT COUNT(*) FROM tasks
+         WHERE parent_task_id IS NULL
+           AND root_task_id IS NULL
+           AND delegation_depth > 0
+        """
+    )
+    orphans = cur.fetchone()[0]
+
+    if orphans == 0:
+        print("  No orphan rows found. DB is coherent; nothing to do.")
+        conn.close()
+        return
+
+    cur.execute(
+        """
+        UPDATE tasks
+           SET delegation_depth = 0,
+               status = CASE
+                 WHEN status = 'delegated' THEN 'open'
+                 ELSE status
+               END,
+               updated_at = datetime('now')
+         WHERE parent_task_id IS NULL
+           AND root_task_id IS NULL
+           AND delegation_depth > 0
+        """
+    )
+    conn.commit()
+    conn.close()
+    print(f"  Normalized {orphans} orphan row(s): delegation_depth → 0, status 'delegated' → 'open'.")
+    print("Migration 006 complete.")
+
+
+if __name__ == "__main__":
+    default_path = os.path.join(
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+        "data", "delega.db",
+    )
+    db_path = os.environ.get("DELEGA_DB_PATH", default_path)
+    print(f"Running migration on: {db_path}")
+    migrate(db_path)

backend/tests/test_task_coherence.py

@@ -0,0 +1,215 @@
+"""Tests for delegation_depth coherence guards on the tasks endpoints.
+
+Parity with the hosted API's coherence fix (delega-dev/delega-api#28):
+chain fields (parent_task_id, root_task_id, delegation_depth) must be
+managed atomically by the server. Clients cannot set them directly on
+POST /tasks or mutate them via PUT /tasks/:id. Chains are built only
+via POST /tasks/:id/delegate, which sets all three atomically.
+"""
+
+import os
+import sys
+import pytest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+os.environ["DELEGA_REQUIRE_AUTH"] = "true"
+os.environ["DELEGA_DB_PATH"] = ":memory:"
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from fastapi.testclient import TestClient
+
+import main
+import models
+from main import app, get_db, derive_key_lookup, derive_key_hash
+
+
+# ---------- Fixtures (mirrors test_permissions.py pattern) ----------
+
+@pytest.fixture(autouse=True)
+def fresh_db(tmp_path):
+    db_path = str(tmp_path / "test.db")
+    test_engine = create_engine(
+        f"sqlite:///{db_path}",
+        connect_args={"check_same_thread": False},
+    )
+    models.Base.metadata.create_all(bind=test_engine)
+    TestSession = sessionmaker(autocommit=False, autoflush=False, bind=test_engine)
+
+    def override_get_db():
+        db = TestSession()
+        try:
+            yield db
+        finally:
+            db.close()
+
+    app.dependency_overrides[get_db] = override_get_db
+    _orig_session_local = main.SessionLocal
+    main.SessionLocal = TestSession
+    main._rate_limiter._hits.clear()
+
+    yield test_engine
+
+    main.SessionLocal = _orig_session_local
+    app.dependency_overrides.clear()
+
+
+@pytest.fixture()
+def client():
+    return TestClient(app, base_url="http://localhost")
+
+
+def make_agent(engine, name: str, *, is_admin: bool = False):
+    import secrets as _secrets
+    raw_key = "dlg_" + _secrets.token_urlsafe(32)
+    salt = _secrets.token_hex(16)
+    key_hash = derive_key_hash(raw_key, salt)
+    key_lookup = derive_key_lookup(raw_key)
+
+    Session = sessionmaker(bind=engine)
+    db = Session()
+    agent = models.Agent(
+        name=name,
+        api_key=raw_key,
+        key_hash=key_hash,
+        key_salt=salt,
+        key_lookup=key_lookup,
+        is_admin=is_admin,
+        permissions=[],
+        active=True,
+    )
+    db.add(agent)
+    db.commit()
+    db.refresh(agent)
+    agent_id = agent.id
+    db.close()
+    return agent_id, raw_key
+
+
+def auth(key: str):
+    return {"X-Agent-Key": key}
+
+
+# ---------- POST /tasks coherence ----------
+
+class TestCreateTaskCoherence:
+    def test_bare_create_defaults_to_depth_zero_root_self(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        r = client.post("/api/tasks", json={"content": "hello"}, headers=auth(key))
+        assert r.status_code == 200
+        body = r.json()
+        assert body["delegation_depth"] == 0
+        assert body["parent_task_id"] is None
+        assert body["root_task_id"] == body["id"]
+
+    def test_explicit_zero_depth_accepted(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        r = client.post(
+            "/api/tasks",
+            json={"content": "explicit zero", "delegation_depth": 0},
+            headers=auth(key),
+        )
+        assert r.status_code == 200
+        assert r.json()["delegation_depth"] == 0
+
+    def test_non_zero_depth_rejected_400(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        r = client.post(
+            "/api/tasks",
+            json={"content": "orphan attempt", "delegation_depth": 3},
+            headers=auth(key),
+        )
+        assert r.status_code == 400
+        assert "delegation_depth" in r.json()["detail"]
+
+    def test_root_task_id_in_body_rejected_400(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        r = client.post(
+            "/api/tasks",
+            json={"content": "x", "root_task_id": 99},
+            headers=auth(key),
+        )
+        assert r.status_code == 400
+        assert "root_task_id" in r.json()["detail"]
+
+    def test_parent_task_id_still_accepted(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        parent = client.post("/api/tasks", json={"content": "parent"}, headers=auth(key)).json()
+        r = client.post(
+            "/api/tasks",
+            json={"content": "child", "parent_task_id": parent["id"]},
+            headers=auth(key),
+        )
+        assert r.status_code == 200
+        child = r.json()
+        assert child["parent_task_id"] == parent["id"]
+        assert child["root_task_id"] == parent["id"]
+        assert child["delegation_depth"] == 1
+
+
+# ---------- PUT /tasks/:id coherence ----------
+
+class TestUpdateTaskCoherence:
+    def test_delegation_depth_in_put_rejected_400(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        task = client.post("/api/tasks", json={"content": "x"}, headers=auth(key)).json()
+        r = client.put(
+            f"/api/tasks/{task['id']}",
+            json={"delegation_depth": 9},
+            headers=auth(key),
+        )
+        assert r.status_code == 400
+        assert "delegation_depth" in r.json()["detail"]
+
+    def test_parent_task_id_in_put_rejected_400(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        task = client.post("/api/tasks", json={"content": "x"}, headers=auth(key)).json()
+        r = client.put(
+            f"/api/tasks/{task['id']}",
+            json={"parent_task_id": 42},
+            headers=auth(key),
+        )
+        assert r.status_code == 400
+        assert "parent_task_id" in r.json()["detail"]
+
+    def test_root_task_id_in_put_rejected_400(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        task = client.post("/api/tasks", json={"content": "x"}, headers=auth(key)).json()
+        r = client.put(
+            f"/api/tasks/{task['id']}",
+            json={"root_task_id": 7},
+            headers=auth(key),
+        )
+        assert r.status_code == 400
+        assert "root_task_id" in r.json()["detail"]
+
+    def test_regular_content_update_still_works(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        task = client.post("/api/tasks", json={"content": "old"}, headers=auth(key)).json()
+        r = client.put(
+            f"/api/tasks/{task['id']}",
+            json={"content": "new"},
+            headers=auth(key),
+        )
+        assert r.status_code == 200
+        assert r.json()["content"] == "new"
+
+
+# ---------- POST /tasks/:id/delegate atomicity ----------
+
+class TestDelegateAtomicity:
+    def test_delegate_sets_chain_fields_atomically(self, fresh_db, client):
+        _, key = make_agent(fresh_db, "a", is_admin=True)
+        parent = client.post("/api/tasks", json={"content": "root"}, headers=auth(key)).json()
+        r = client.post(
+            f"/api/tasks/{parent['id']}/delegate",
+            json={"content": "delegated"},
+            headers=auth(key),
+        )
+        assert r.status_code == 200
+        child = r.json()
+        assert child["parent_task_id"] == parent["id"]
+        assert child["root_task_id"] == parent["id"]
+        assert child["delegation_depth"] == 1