Add fail-closed behavior link switching

server/API_V2_REFERENCE.md

@@ -2464,6 +2464,43 @@ Source code is **not** set here. After linking, upload the code with
 
 **Response** `201 Created` — returns the behavior link (same shape as list items above).
 
+#### Create a Fresh Behavior Definition
+
+```http
+POST /v2/companions/{companion_id}/behavior-definitions
+Content-Type: application/json
+
+{
+  "key": "relationship_update_step06_v1",
+  "name": "Relationship update v1",
+  "source_code": "async def execute(ctx): return None"
+}
+```
+
+This creates a new project-scoped definition and never links or updates an
+existing definition. An existing project key returns `409 Conflict`.
+
+#### Switch a Behavior Definition
+
+```http
+POST /v2/companions/{companion_id}/behaviors/{behavior_key}/switch
+Content-Type: application/json
+
+{
+  "behavior_key": "relationship_update",
+  "new_behavior_id": "...",
+  "links": [
+    {"link_id": "...", "relationship_id": null, "expected_behavior_id": "..."},
+    {"link_id": "...", "relationship_id": "...", "expected_behavior_id": "..."}
+  ]
+}
+```
+
+This endpoint currently returns `503 Service Unavailable` and performs no
+mutation. The runtime has no distributed execution gate, in-flight job drain,
+or cross-worker cache invalidation contract. The repository's transactional
+compare-and-swap is not a live cutover API until those gates are implemented.
+
 ---
 
 #### Update Behavior Definition (source code)

server/app/models/v2/__init__.py

@@ -8,9 +8,12 @@
     BehaviorLink,
     BehaviorLinkCreate,
     BehaviorLinkResponse,
+    BehaviorLinkSwitchRequest,
+    BehaviorLinkSwitchTarget,
     BehaviorLinkUpdate,
     BehaviorResponse,
     BehaviorUpdate,
+    IsolatedBehaviorDefinitionCreate,
     CompanionBehaviorsResponse,
     RelationshipBehaviorsResponse,
     format_trigger_shorthand,
@@ -51,9 +54,12 @@
     "BehaviorLink",
     "BehaviorLinkCreate",
     "BehaviorLinkResponse",
+    "BehaviorLinkSwitchRequest",
+    "BehaviorLinkSwitchTarget",
     "BehaviorLinkUpdate",
     "BehaviorResponse",
     "BehaviorUpdate",
+    "IsolatedBehaviorDefinitionCreate",
     "CompanionBehaviorsResponse",
     # Message
     "MessageCreate",

server/app/models/v2/behavior.py

@@ -239,6 +239,38 @@ class BehaviorLinkUpdate(BaseModel):
     params: Dict[str, Any] | None = None
 
 
+class BehaviorLinkSwitchTarget(BaseModel):
+    """One explicitly scoped default or relationship override link."""
+
+    link_id: UUID
+    relationship_id: UUID | None = None
+    expected_behavior_id: UUID
+
+
+class IsolatedBehaviorDefinitionCreate(BaseModel):
+    """Create a fresh project definition without reusing an existing key."""
+
+    key: str = Field(..., min_length=1)
+    name: str = Field(..., min_length=1)
+    description: str | None = None
+    source_code: str = ""
+    dependencies: List[str] = Field(default_factory=list)
+    timeout_seconds: int = Field(60, ge=1, le=600)
+    block_network: bool = True
+
+
+class BehaviorLinkSwitchRequest(BaseModel):
+    """Compare-and-swap request for a companion behavior link.
+
+    Relationship override links must be named explicitly. The server rejects
+    incomplete or stale scopes instead of guessing which overrides to move.
+    """
+
+    behavior_key: str = Field(..., min_length=1)
+    new_behavior_id: UUID
+    links: List[BehaviorLinkSwitchTarget] = Field(..., min_length=1)
+
+
 class BehaviorLink(BehaviorLinkBase):
     """Full behavior link model."""
 

server/app/repositories/behavior_repository.py

@@ -1041,3 +1041,139 @@ async def disable_behavior_link(
             )
             return True
         return False
+
+    @staticmethod
+    async def switch_companion_behavior_links(
+        conn: asyncpg.Connection,
+        *,
+        companion_id: UUID,
+        project_id: UUID,
+        behavior_key: str,
+        new_behavior_id: UUID,
+        targets: List[Dict[str, Any]],
+    ) -> Dict[str, Any]:
+        """Atomically compare-and-swap an exact companion behavior link set.
+
+        This transaction does not drain runtime work or notify remote caches;
+        callers must keep live switching disabled until those guarantees exist.
+        """
+        requested: dict[UUID, Dict[str, Any]] = {}
+        for target in targets:
+            link_id = target["link_id"]
+            if link_id in requested:
+                raise ValueError("duplicate target link_id")
+            if target["expected_behavior_id"] == new_behavior_id:
+                raise ValueError("expected behavior_id must differ from the new definition")
+            requested[link_id] = target
+        if not requested:
+            raise ValueError("at least one target link is required")
+
+        relationship_ids: set[UUID] = set()
+        async with conn.transaction():
+            # Existing API link writers take ROW EXCLUSIVE locks. Serialize the
+            # inventory and swap with them so the override set cannot drift.
+            await conn.execute(
+                "LOCK TABLE companion_behavior_links IN SHARE ROW EXCLUSIVE MODE"
+            )
+            companion = await conn.fetchrow(
+                "SELECT project_id FROM companions WHERE id = $1 FOR SHARE",
+                companion_id,
+            )
+            if not companion or companion["project_id"] != project_id:
+                raise ValueError("companion is missing or outside the project")
+
+            new_definition = await conn.fetchrow(
+                "SELECT id, project_id, key FROM behaviors WHERE id = $1 FOR SHARE",
+                new_behavior_id,
+            )
+            if not new_definition or new_definition["project_id"] != project_id:
+                raise ValueError("new behavior definition is missing or outside the project")
+            if new_definition["key"] == behavior_key:
+                raise ValueError("new definition must use a fresh behavior key")
+
+            actual_rows = await conn.fetch(
+                """
+                SELECT cbl.id AS link_id, cbl.companion_id, cbl.relationship_id,
+                       cbl.behavior_id, b.key AS behavior_key,
+                       r.companion_id AS relationship_companion_id
+                FROM companion_behavior_links cbl
+                JOIN behaviors b ON b.id = cbl.behavior_id
+                LEFT JOIN relationships r ON r.id = cbl.relationship_id
+                WHERE cbl.companion_id = $1
+                  AND b.key IN ($2, $3)
+                ORDER BY cbl.id
+                FOR UPDATE OF cbl
+                """,
+                companion_id,
+                behavior_key,
+                new_definition["key"],
+            )
+            actual_by_id = {row["link_id"]: row for row in actual_rows}
+            if set(actual_by_id) != set(requested):
+                raise ValueError("target link set is incomplete or contains non-target links")
+
+            already_switched = True
+            switched_link_ids = set()
+            for link_id, target in requested.items():
+                row = actual_by_id[link_id]
+                if row["companion_id"] != companion_id:
+                    raise ValueError("target link belongs to a different companion")
+                if row["relationship_id"] != target["relationship_id"]:
+                    raise ValueError("target relationship scope changed")
+                if (
+                    row["relationship_id"] is not None
+                    and row["relationship_companion_id"] != companion_id
+                ):
+                    raise ValueError("relationship override belongs to another companion")
+                if row["behavior_id"] != target["expected_behavior_id"]:
+                    if row["behavior_id"] != new_behavior_id:
+                        raise ValueError("expected behavior_id changed")
+                    switched_link_ids.add(link_id)
+                else:
+                    already_switched = False
+                if row["relationship_id"] is not None:
+                    relationship_ids.add(row["relationship_id"])
+
+            if switched_link_ids and not already_switched:
+                raise ValueError("expected behavior_id changed")
+
+            if not already_switched:
+                active_jobs = await conn.fetchval(
+                    """
+                    SELECT EXISTS (
+                        SELECT 1 FROM jobs
+                        WHERE companion_id = $1
+                          AND behavior_key = $2
+                          AND status IN ('pending', 'claimed', 'running')
+                    )
+                    """,
+                    companion_id,
+                    behavior_key,
+                )
+                if active_jobs:
+                    raise ValueError("active behavior jobs must be drained before switching")
+
+                for link_id, target in requested.items():
+                    if actual_by_id[link_id]["behavior_id"] == new_behavior_id:
+                        continue
+                    changed = await conn.fetchval(
+                        """
+                        UPDATE companion_behavior_links
+                        SET behavior_id = $1, updated_at = NOW()
+                        WHERE id = $2 AND companion_id = $3 AND behavior_id = $4
+                        RETURNING id
+                        """,
+                        new_behavior_id,
+                        link_id,
+                        companion_id,
+                        target["expected_behavior_id"],
+                    )
+                    if changed is None:
+                        raise ValueError("link compare-and-swap failed")
+
+        # Cache invalidation is process-local. It is useful for local callers,
+        # but not sufficient as a live cutover protocol across worker replicas.
+        BehaviorRepository.invalidate_behavior_cache(companion_id)
+        for relationship_id in relationship_ids:
+            BehaviorRepository.invalidate_behavior_cache(companion_id, relationship_id)
+        return {"switched": len(requested), "already_switched": already_switched}

server/app/routers/v2/behaviors.py

@@ -30,10 +30,12 @@
 from ...models.v2.behavior import (
     BehaviorLinkCreate,
     BehaviorLinkResponse,
+    BehaviorLinkSwitchRequest,
     BehaviorLinkUpdate,
     BehaviorResponse,
     BehaviorUpdate,
     CompanionBehaviorsResponse,
+    IsolatedBehaviorDefinitionCreate,
     RelationshipBehaviorsResponse,
     TriggerBehaviorRequest,
     TriggerBehaviorResponse,
@@ -141,6 +143,79 @@ def _to_behavior_response(behavior: Dict[str, Any]) -> BehaviorResponse:
 # -----------------------------------------------------------------------------
 
 
+@router.post(
+    "/companions/{companion_id}/behavior-definitions",
+    response_model=BehaviorResponse,
+    status_code=status.HTTP_201_CREATED,
+    summary="Create a fresh behavior definition",
+    description="Create a new project-scoped definition; existing keys are never reused.",
+)
+async def create_isolated_behavior_definition(
+    companion_id: UUID,
+    body: IsolatedBehaviorDefinitionCreate,
+    subject: ProjectApiKeySubject = Depends(get_project_api_subject),
+    conn: asyncpg.Connection = Depends(get_db),
+):
+    """Create a fresh definition for a companion's later explicit link switch."""
+    await _verify_companion_access(conn, companion_id, subject.project.id)
+    existing = await BehaviorRepository.get_behavior_by_project_key(
+        conn, project_id=subject.project.id, behavior_key=body.key
+    )
+    if existing:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Behavior key already exists; definition reuse is not allowed here",
+        )
+    try:
+        async with conn.transaction():
+            created = await BehaviorRepository.create_behavior(
+                conn,
+                behavior_id=uuid4(),
+                project_id=subject.project.id,
+                key=body.key,
+                name=body.name,
+                description=body.description,
+                source_code=body.source_code,
+                dependencies=body.dependencies,
+                timeout_seconds=body.timeout_seconds,
+                block_network=body.block_network,
+            )
+    except asyncpg.UniqueViolationError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Behavior key already exists; definition reuse is not allowed here",
+        ) from exc
+    return _to_behavior_response(created)
+
+
+@router.post(
+    "/companions/{companion_id}/behaviors/{behavior_key}/switch",
+    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+    summary="Switch behavior definition (disabled until runtime drain is supported)",
+)
+async def switch_companion_behavior_definition(
+    companion_id: UUID,
+    behavior_key: str,
+    body: BehaviorLinkSwitchRequest,
+    subject: ProjectApiKeySubject = Depends(get_project_api_subject),
+    conn: asyncpg.Connection = Depends(get_db),
+):
+    """Fail closed until distributed execution drain/cache invalidation exists."""
+    await _verify_companion_access(conn, companion_id, subject.project.id)
+    if body.behavior_key != behavior_key:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Request behavior_key must match the URL",
+        )
+    raise HTTPException(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        detail=(
+            "Live behavior switching is disabled: the runtime cannot yet block new executions, "
+            "confirm all in-flight jobs are drained, or invalidate caches across worker processes"
+        ),
+    )
+
+
 @router.get(
     "/companions/{companion_id}/behaviors",
     response_model=CompanionBehaviorsResponse,