fix: enforce unique governed dataflow names

.superpowers/sdd/task-8-report.md

@@ -358,11 +358,10 @@ Fifth-review acceptance:
 
 - focused Saga/repository/cutover contracts: `40 passed`;
 - real PostgreSQL/API/Neo4j integration: `3 passed`;
-- full repository run completed with
-  `680 passed, 33 skipped, 59 subtests passed` and one unrelated existing
-  publication-receipt assertion mismatch: the tampered receipt was correctly
-  rejected as non-canonical encoding while that randomized test expected the
-  word `signature`;
+- a full repository run exposed one overly narrow receipt-tamper assertion:
+  modified base64url input can correctly fail either canonical encoding or
+  signature verification. The assertion now accepts both fail-closed reasons;
+  the fresh final run is `681 passed, 33 skipped, 59 subtests passed`;
 - selected Ruff `F`, `I`, and `B` checks passed;
 - the production frontend build completed with zero errors and the same
   existing 20 console warnings;
@@ -371,6 +370,43 @@ Fifth-review acceptance:
   health reports database and Neo4j healthy with code 200, and frontend HTTP
   returns 200.
 
+## DataFlow name concurrency closeout
+
+The sixth review closed the final governed-create time-of-check/time-of-use
+window around `DataFlow.name_zh`:
+
+- governed graph creation now installs and consumes both
+  `data_flow_uid` and `data_flow_name_zh` uniqueness constraints before its
+  preflight and UID-keyed `MERGE`;
+- constraint installation scans existing graph state. Historical duplicate
+  UIDs or Chinese names make installation fail closed; they are never silently
+  accepted or rewritten;
+- concurrent first-time schema installation retries only Neo4j deadlock
+  victims. Constraint validation/verification/creation failures and concurrent
+  unique-property violations are normalized to `dataflow_uid_conflict`;
+- the existing same-name preflight remains useful diagnostics, while the
+  database uniqueness constraint is authoritative at the concurrent write
+  boundary.
+
+Sixth-review acceptance:
+
+- focused Saga/repository/publication/cutover contracts: `47 passed`;
+- the dedicated real-Neo4j integration suite is `2 passed`: two concurrent
+  creates with different UUIDv7 values and the same `name_zh` produced exactly
+  one success and one `dataflow_uid_conflict`, left one node, and verified both
+  constraints;
+- a simulated historical duplicate made name-constraint installation fail
+  before any graph `MERGE`;
+- full repository suite:
+  `682 passed, 34 skipped, 59 subtests passed`;
+- selected Ruff `F`, `I`, and `B` checks and whitespace validation passed;
+- the production frontend build completed with zero errors and the same
+  existing 20 console warnings;
+- the final backend image was rebuilt from the reviewed source; backend and
+  frontend are healthy, Alembic remains `20260724_230 (head)`, application
+  health reports database and Neo4j healthy with code 200, and frontend HTTP
+  returns 200.
+
 ## Residual scope
 
 - Production-line cross-station compatibility remains ultimately authoritative

app/core/data_flow/dataflows.py

@@ -154,6 +154,42 @@ class DataFlowService:
             "migration_metadata": copy.deepcopy(metadata),
         }
 
+    @staticmethod
+    def _is_dataflow_constraint_conflict(error: Exception) -> bool:
+        code = str(getattr(error, "code", "") or "")
+        return code in {
+            "Neo.ClientError.Schema.ConstraintCreationFailed",
+            "Neo.ClientError.Schema.ConstraintValidationFailed",
+            "Neo.ClientError.Schema.ConstraintVerificationFailed",
+        }
+
+    @staticmethod
+    def _install_dataflow_constraints(session) -> None:
+        """Install both identity constraints, retrying only schema deadlocks."""
+        for attempt in range(3):
+            try:
+                session.run(
+                    "CREATE CONSTRAINT data_flow_uid IF NOT EXISTS "
+                    "FOR (n:DataFlow) REQUIRE n.uid IS UNIQUE"
+                ).consume()
+                session.run(
+                    "CREATE CONSTRAINT data_flow_name_zh IF NOT EXISTS "
+                    "FOR (n:DataFlow) REQUIRE n.name_zh IS UNIQUE"
+                ).consume()
+                return
+            except Exception as exc:
+                if DataFlowService._is_dataflow_constraint_conflict(exc):
+                    raise ValueError("dataflow_uid_conflict") from exc
+                code = str(getattr(exc, "code", "") or "")
+                if (
+                    code
+                    == "Neo.TransientError.Transaction.DeadlockDetected"
+                    and attempt < 2
+                ):
+                    continue
+                raise
+        raise RuntimeError("DataFlow constraints were not installed")
+
     @staticmethod
     def _merge_governed_dataflow(node_data: dict[str, Any]) -> tuple[int, dict]:
         """Idempotently create by stable UID and reject immutable conflicts."""
@@ -167,10 +203,7 @@ class DataFlowService:
         driver = connect_graph()
         try:
             with driver.session() as session:
-                session.run(
-                    "CREATE CONSTRAINT data_flow_uid IF NOT EXISTS "
-                    "FOR (n:DataFlow) REQUIRE n.uid IS UNIQUE"
-                )
+                DataFlowService._install_dataflow_constraints(session)
                 conflict = session.run(
                     "MATCH (n:DataFlow {name_zh: $name_zh}) "
                     "WHERE n.uid IS NULL OR n.uid <> $uid "
@@ -179,12 +212,17 @@ class DataFlowService:
                 ).single()
                 if conflict is not None:
                     raise ValueError("dataflow_uid_conflict")
-                record = session.run(
-                    "MERGE (n:DataFlow {uid: $uid}) "
-                    "ON CREATE SET n = $properties "
-                    "RETURN n, id(n) AS node_id",
-                    {"uid": node_data["uid"], "properties": node_data},
-                ).single()
+                try:
+                    record = session.run(
+                        "MERGE (n:DataFlow {uid: $uid}) "
+                        "ON CREATE SET n = $properties "
+                        "RETURN n, id(n) AS node_id",
+                        {"uid": node_data["uid"], "properties": node_data},
+                    ).single()
+                except Exception as exc:
+                    if DataFlowService._is_dataflow_constraint_conflict(exc):
+                        raise ValueError("dataflow_uid_conflict") from exc
+                    raise
                 if record is None:
                     raise RuntimeError(
                         "governed DataFlow MERGE returned no node"

tests/core/data_rules/test_publication.py

@@ -229,7 +229,11 @@ def test_generation_receipt_rejects_expired_and_modified_signatures():
         )
 
     valid = signer.issue(_receipt_claims(spec, actor))
-    with pytest.raises(ValueError, match="signature"):
+    # A one-character base64url mutation may either preserve canonical
+    # encoding and fail the signature check, or alter only padding bits and
+    # fail the earlier canonical-encoding check. Both are required
+    # fail-closed outcomes for a modified receipt.
+    with pytest.raises(ValueError, match=r"receipt (signature|encoding)"):
         signer.verify(
             valid[:-1] + ("A" if valid[-1] != "A" else "B"),
             actor_uid=actor,

tests/integration/test_dataflow_create_saga_neo4j.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import os
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier
 
 import pytest
 from neo4j import GraphDatabase
@@ -10,6 +12,76 @@ from app.core.common.identifiers import new_governance_uid
 from app.core.data_flow.dataflows import DataFlowService
 
 
+def test_real_neo4j_concurrent_different_uids_same_name_is_closed():
+    uri = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_URI")
+    password = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_PASSWORD")
+    if not uri or not password:
+        pytest.skip("real Neo4j acceptance connection is not configured")
+    user = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_USER", "neo4j")
+    app = create_app()
+    app.config.update(
+        TESTING=True,
+        NEO4J_URI=uri,
+        NEO4J_USER=user,
+        NEO4J_PASSWORD=password,
+        NEO4J_ENCRYPTED=False,
+    )
+    name = f"同名并发生产线-{new_governance_uid()}"
+    uids = [new_governance_uid(), new_governance_uid()]
+    start = Barrier(2)
+    driver = GraphDatabase.driver(
+        uri, auth=(user, password), encrypted=False
+    )
+
+    def create(uid):
+        start.wait()
+        try:
+            with app.app_context():
+                DataFlowService._merge_governed_dataflow(
+                    {
+                        "uid": uid,
+                        "name_zh": name,
+                        "name_en": f"concurrent_{uid.replace('-', '')}",
+                        "script_type": "governed",
+                        "script_requirement": "{}",
+                        "script_path": "",
+                    }
+                )
+            return "created"
+        except ValueError as exc:
+            return str(exc)
+
+    try:
+        with driver.session() as session:
+            session.run(
+                "MATCH (n:DataFlow {name_zh: $name}) DETACH DELETE n",
+                {"name": name},
+            ).consume()
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            outcomes = list(pool.map(create, uids))
+        assert sorted(outcomes) == ["created", "dataflow_uid_conflict"]
+        with driver.session() as session:
+            count = session.run(
+                "MATCH (n:DataFlow {name_zh: $name}) "
+                "RETURN count(n) AS count",
+                {"name": name},
+            ).single()["count"]
+            constraints = session.run(
+                "SHOW CONSTRAINTS YIELD name "
+                "WHERE name IN ['data_flow_uid', 'data_flow_name_zh'] "
+                "RETURN collect(name) AS names"
+            ).single()["names"]
+        assert count == 1
+        assert set(constraints) == {"data_flow_uid", "data_flow_name_zh"}
+    finally:
+        with driver.session() as session:
+            session.run(
+                "MATCH (n:DataFlow {name_zh: $name}) DETACH DELETE n",
+                {"name": name},
+            ).consume()
+        driver.close()
+
+
 def test_real_neo4j_uid_constraint_merge_replay_and_conflict():
     uri = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_URI")
     password = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_PASSWORD")
@@ -52,12 +124,13 @@ def test_real_neo4j_uid_constraint_merge_replay_and_conflict():
             constraints = [
                 record["name"]
                 for record in session.run(
-                    "SHOW CONSTRAINTS YIELD name WHERE name = 'data_flow_uid' "
+                    "SHOW CONSTRAINTS YIELD name "
+                    "WHERE name IN ['data_flow_uid', 'data_flow_name_zh'] "
                     "RETURN name"
                 )
             ]
         assert count == 1
-        assert constraints == ["data_flow_uid"]
+        assert set(constraints) == {"data_flow_uid", "data_flow_name_zh"}
     finally:
         with driver.session() as session:
             session.run("MATCH (n:DataFlow {uid: $uid}) DETACH DELETE n", {"uid": uid})

tests/test_dataflow_create_saga.py

@@ -16,17 +16,28 @@ class GraphResult:
     def single(self):
         return self.record
 
+    def consume(self):
+        return None
+
 
 class GraphSession:
-    def __init__(self, *, existing=None, conflict=False):
+    def __init__(
+        self, *, existing=None, conflict=False, constraint_error=None
+    ):
         self.existing = existing
         self.conflict = conflict
+        self.constraint_error = constraint_error
         self.calls = []
 
     def run(self, query, parameters=None, **kwargs):
         values = parameters or kwargs
         self.calls.append((query, values))
         if query.startswith("CREATE CONSTRAINT"):
+            if (
+                self.constraint_error is not None
+                and "data_flow_name_zh" in query
+            ):
+                raise self.constraint_error
             return GraphResult()
         if "WHERE n.uid IS NULL OR n.uid <> $uid" in query:
             return GraphResult({"uid": "other"}) if self.conflict else GraphResult()
@@ -86,6 +97,12 @@ def test_governed_graph_create_installs_constraint_and_reconciles_same_uid(
         "FOR (n:DataFlow) REQUIRE n.uid IS UNIQUE"
         for call in graph.calls
     )
+    assert any(
+        call[0]
+        == "CREATE CONSTRAINT data_flow_name_zh IF NOT EXISTS "
+        "FOR (n:DataFlow) REQUIRE n.name_zh IS UNIQUE"
+        for call in graph.calls
+    )
     assert sum(call[0].startswith("MERGE") for call in graph.calls) == 2
 
 
@@ -110,6 +127,23 @@ def test_governed_graph_reconcile_fails_closed_on_uid_or_name_conflict(
         DataFlowService._merge_governed_dataflow(node)
 
 
+def test_governed_graph_fails_closed_when_name_constraint_finds_duplicates(
+    monkeypatch,
+):
+    class ConstraintFailure(RuntimeError):
+        code = "Neo.ClientError.Schema.ConstraintValidationFailed"
+
+    graph = GraphSession(constraint_error=ConstraintFailure("duplicates"))
+    monkeypatch.setattr(
+        "app.core.data_flow.dataflows.connect_graph",
+        lambda: GraphDriver(graph),
+    )
+
+    with pytest.raises(ValueError, match="dataflow_uid_conflict"):
+        DataFlowService._merge_governed_dataflow(governed_node())
+    assert not any(call[0].startswith("MERGE") for call in graph.calls)
+
+
 def test_completed_saga_replays_result_without_another_neo4j_write(monkeypatch):
     expected = {"id": 73, **governed_node()}