diff --git a/server/API_V2_REFERENCE.md b/server/API_V2_REFERENCE.md index 77626c5..a68f4eb 100644 --- a/server/API_V2_REFERENCE.md +++ b/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) diff --git a/server/app/models/v2/__init__.py b/server/app/models/v2/__init__.py index fe02c24..b5f392d 100644 --- a/server/app/models/v2/__init__.py +++ b/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", diff --git a/server/app/models/v2/behavior.py b/server/app/models/v2/behavior.py index 36f3ff7..0312041 100644 --- a/server/app/models/v2/behavior.py +++ b/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.""" diff --git a/server/app/repositories/behavior_repository.py b/server/app/repositories/behavior_repository.py index 1f8305a..904a167 100644 --- a/server/app/repositories/behavior_repository.py +++ b/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} diff --git a/server/app/routers/v2/behaviors.py b/server/app/routers/v2/behaviors.py index 0203544..bbfe76e 100644 --- a/server/app/routers/v2/behaviors.py +++ b/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, diff --git a/server/tests/test_behavior_safe_switch.py b/server/tests/test_behavior_safe_switch.py new file mode 100644 index 0000000..e71dfab --- /dev/null +++ b/server/tests/test_behavior_safe_switch.py @@ -0,0 +1,311 @@ +"""Synthetic database contract tests for behavior link compare-and-swap.""" + +from __future__ import annotations + +import asyncio +import os +from uuid import UUID, uuid4 + +import asyncpg +import pytest + +from app.repositories.behavior_repository import BehaviorRepository + + +@pytest.fixture +async def synthetic_conn(): + """Use an explicitly designated disposable database, never a dev/live DSN.""" + dsn = os.getenv("PM_SAFE_SWITCH_TEST_DSN") + if not dsn: + pytest.skip("PM_SAFE_SWITCH_TEST_DSN is not configured") + conn = await asyncpg.connect(dsn) + try: + yield conn + finally: + await conn.close() + + +def _id(value: int) -> UUID: + return UUID(int=value) + + +async def _seed(conn, *, temporary_tables: bool = True): + project_id = _id(1) + companion_ids = [_id(i) for i in range(10, 14)] + old_behavior_id = _id(20) + new_behavior_id = _id(21) + relationship_ids = [_id(i) for i in range(30, 34)] + link_ids = [_id(i) for i in range(40, 48)] + + temp = "TEMP " if temporary_tables else "" + await conn.execute( + f"CREATE {temp}TABLE companions (id UUID PRIMARY KEY, project_id UUID NOT NULL)" + ) + await conn.execute( + f"CREATE {temp}TABLE behaviors (id UUID PRIMARY KEY, project_id UUID NOT NULL, " + "key TEXT UNIQUE NOT NULL, source_code TEXT NOT NULL)" + ) + await conn.execute( + f"CREATE {temp}TABLE relationships (id UUID PRIMARY KEY, companion_id UUID NOT NULL)" + ) + await conn.execute( + f"CREATE {temp}TABLE companion_behavior_links (id UUID PRIMARY KEY, " + "companion_id UUID NOT NULL, " + "relationship_id UUID NULL, behavior_id UUID NOT NULL REFERENCES behaviors(id), " + "updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW())" + ) + await conn.execute( + f"CREATE {temp}TABLE jobs (companion_id UUID, behavior_key TEXT, status TEXT NOT NULL)" + ) + await conn.execute( + "INSERT INTO behaviors VALUES ($1, $2, $3, $4)", + old_behavior_id, + project_id, + "old", + "keep-me", + ) + await conn.execute( + "INSERT INTO behaviors VALUES ($1, $2, $3, $4)", + new_behavior_id, + project_id, + "new-v1", + "new-source", + ) + + for index, companion_id in enumerate(companion_ids): + await conn.execute("INSERT INTO companions VALUES ($1, $2)", companion_id, project_id) + await conn.execute( + "INSERT INTO relationships VALUES ($1, $2)", + relationship_ids[index], + companion_id, + ) + await conn.execute( + "INSERT INTO companion_behavior_links " + "(id, companion_id, behavior_id) VALUES ($1, $2, $3)", + link_ids[index], + companion_id, + old_behavior_id, + ) + await conn.execute( + "INSERT INTO companion_behavior_links (id, companion_id, relationship_id, behavior_id) " + "VALUES ($1, $2, $3, $4)", + link_ids[index + 4], + companion_id, + relationship_ids[index], + old_behavior_id, + ) + + targets = [ + { + "link_id": link_ids[0], + "relationship_id": None, + "expected_behavior_id": old_behavior_id, + }, + { + "link_id": link_ids[4], + "relationship_id": relationship_ids[0], + "expected_behavior_id": old_behavior_id, + }, + ] + return project_id, companion_ids, old_behavior_id, new_behavior_id, targets, link_ids + + +@pytest.mark.live +async def test_safe_switch_changes_only_explicit_companion_and_is_idempotent(synthetic_conn): + conn = synthetic_conn + project_id, companions, old_id, new_id, targets, links = await _seed(conn) + + result = await BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets, + ) + assert result == {"switched": 2, "already_switched": False} + retry = await BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets, + ) + assert retry == {"switched": 2, "already_switched": True} + + target_rows = await conn.fetch( + "SELECT behavior_id FROM companion_behavior_links WHERE id = ANY($1::uuid[])", + [links[0], links[4]], + ) + assert {row["behavior_id"] for row in target_rows} == {new_id} + other_rows = await conn.fetch( + "SELECT behavior_id FROM companion_behavior_links WHERE companion_id = ANY($1::uuid[])", + companions[1:], + ) + assert len(other_rows) == 6 + assert all(row["behavior_id"] == old_id for row in other_rows) + definition = await conn.fetchrow("SELECT source_code FROM behaviors WHERE id = $1", old_id) + assert definition["source_code"] == "keep-me" + + +@pytest.mark.live +async def test_safe_switch_rejects_missing_override_stale_expectation_and_active_jobs( + synthetic_conn, +): + conn = synthetic_conn + project_id, companions, old_id, new_id, targets, _links = await _seed(conn) + + with pytest.raises(ValueError, match="incomplete"): + await BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets[:1], + ) + with pytest.raises(ValueError, match="incomplete"): + await BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets + [ + { + "link_id": _id(41), + "relationship_id": None, + "expected_behavior_id": old_id, + } + ], + ) + unchanged = await conn.fetchval( + "SELECT COUNT(*) FROM companion_behavior_links " + "WHERE companion_id = $1 AND behavior_id = $2", + companions[0], + old_id, + ) + assert unchanged == 2 + + await conn.execute( + "UPDATE companion_behavior_links SET behavior_id = $1 WHERE id = $2", + new_id, + targets[0]["link_id"], + ) + with pytest.raises(ValueError, match="expected behavior_id changed"): + await BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets, + ) + + await conn.execute( + "UPDATE companion_behavior_links SET behavior_id = $1 WHERE id = $2", + old_id, + targets[0]["link_id"], + ) + for active_status in ("pending", "claimed", "running"): + await conn.execute("DELETE FROM jobs") + await conn.execute( + "INSERT INTO jobs VALUES ($1, 'old', $2)", companions[0], active_status + ) + with pytest.raises(ValueError, match="active behavior jobs"): + await BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets, + ) + + +@pytest.mark.live +async def test_safe_switch_rolls_back_all_links_when_an_update_fails(synthetic_conn): + conn = synthetic_conn + project_id, companions, old_id, new_id, targets, _links = await _seed(conn) + failing_id = targets[1]["link_id"] + await conn.execute( + f"""CREATE FUNCTION pg_temp.fail_selected_link() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.id = '{failing_id}'::uuid THEN + RAISE EXCEPTION 'synthetic mid-switch failure'; + END IF; + RETURN NEW; + END; $$""" + ) + await conn.execute( + "CREATE TRIGGER fail_selected_link BEFORE UPDATE ON companion_behavior_links " + "FOR EACH ROW EXECUTE FUNCTION pg_temp.fail_selected_link()" + ) + + with pytest.raises(asyncpg.PostgresError, match="synthetic mid-switch failure"): + await BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets, + ) + rows = await conn.fetch( + "SELECT behavior_id FROM companion_behavior_links WHERE companion_id = $1", + companions[0], + ) + assert len(rows) == 2 + assert all(row["behavior_id"] == old_id for row in rows) + + +@pytest.mark.live +async def test_safe_switch_serializes_with_a_concurrent_link_change(synthetic_conn): + conn = synthetic_conn + schema = f"safe_switch_{uuid4().hex}" + await conn.execute(f'CREATE SCHEMA "{schema}"') + other = await asyncpg.connect(os.environ["PM_SAFE_SWITCH_TEST_DSN"]) + try: + await conn.execute(f'SET search_path TO "{schema}"') + await other.execute(f'SET search_path TO "{schema}"') + project_id, companions, old_id, new_id, targets, _links = await _seed( + conn, temporary_tables=False + ) + raced_id = _id(22) + await conn.execute( + "INSERT INTO behaviors VALUES ($1, $2, 'raced', 'raced-source')", + raced_id, + project_id, + ) + + async with other.transaction(): + await other.execute( + "UPDATE companion_behavior_links SET behavior_id = $1 WHERE id = $2", + raced_id, + targets[0]["link_id"], + ) + switch_task = asyncio.create_task( + BehaviorRepository.switch_companion_behavior_links( + conn, + companion_id=companions[0], + project_id=project_id, + behavior_key="old", + new_behavior_id=new_id, + targets=targets, + ) + ) + await asyncio.sleep(0.1) + assert not switch_task.done(), "switch should wait for the concurrent writer" + + with pytest.raises(ValueError, match="incomplete"): + await switch_task + current = await conn.fetchval( + "SELECT behavior_id FROM companion_behavior_links WHERE id = $1", + targets[0]["link_id"], + ) + assert current == raced_id + assert old_id != new_id + finally: + await other.close() + await conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE') diff --git a/web/public/API_V2_REFERENCE.md b/web/public/API_V2_REFERENCE.md index 77626c5..a68f4eb 100644 --- a/web/public/API_V2_REFERENCE.md +++ b/web/public/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)