Authentication and the Sync API

Protect cross-device HTTP routes, make retries idempotent, and converge concurrent session updates under an explicit merge policy.

The browser application so far is local and same-origin. Crossing a device boundary adds identity, authorization headers, token expiry, retries, and concurrent writers. This chapter extends the FastAPI surface with protected sync routes while preserving the SessionRecord and event vocabulary already used by the local REST, WebSocket, CLI, and SQLite layers.

Authentication belongs at the transport boundary

The sync routes require an Authorization: Bearer <token> header. TokenAuthority verifies the HMAC signature, expiry, and revocation before the route parses the session into a sync command. Missing or invalid credentials return 401 with a Bearer challenge. If no authority is configured, the service returns 503 rather than silently exposing sync.

The local browser session routes remain loopback-oriented in this course. A hosted multi-user deployment must put its complete HTTP and WebSocket surface behind managed identity and authorization policy; the small HMAC authority teaches the boundary rather than claiming to replace an identity provider.

from tempfile import TemporaryDirectory

from fastapi.testclient import TestClient

from autocode.domain import SessionRecord
from autocode.runner import DemoAgentRunner
from autocode_service.api import create_app
from autocode_service.auth import TokenAuthority

with TemporaryDirectory() as directory:
    authority = TokenAuthority("course-test-secret")
    token = authority.issue("laptop")
    app = create_app(
        database_path=f"{directory}/sessions.db",
        runner=DemoAgentRunner(),
        token_authority=authority,
    )
    record = SessionRecord("shared", title="Shared session")
    record.append("user_message", {"content": "from laptop"})
    body = {"session": record.to_dict(), "idempotency_key": "laptop:shared:1"}

    with TestClient(app) as client:
        rejected = client.post("/api/sync/sessions", json=body)
        headers = {"Authorization": f"Bearer {token}"}
        accepted = client.post("/api/sync/sessions", json=body, headers=headers)
        pulled = client.get("/api/sync/sessions/shared/events", headers=headers)

assert rejected.status_code == 401
assert accepted.status_code == 200
assert [event["kind"] for event in pulled.json()] == ["user_message"]
print("sync status:", rejected.status_code, "->", accepted.status_code)

sync status: 401 -> 200

The test crosses the real HTTP dependency: no header is rejected, while the issued device token reaches the transport-neutral sync service. The request body carries a serialized session and an idempotency key. A retry with the same key returns the existing merged record instead of adding another logical update.

Conflict policy is part of the API contract

Two devices can append events and update scalar metadata while offline. Append-only events merge by stable event id set union. Scalar fields such as title use deterministic last-writer-wins based on recorded update time. The merge operation must be idempotent, commutative, and associative so delivery order and retry count do not decide the result.

from autocode.domain import SessionRecord
from autocode.sync.conflicts import merge_records

left = SessionRecord("s", title="laptop", updated_at="2026-01-01T00:00:00+00:00")
right = SessionRecord("s", title="desktop", updated_at="2026-01-01T00:00:01+00:00")
left.append("user_message", {"content": "one"})
right.append("user_message", {"content": "two"})
merged = merge_records(left, right)
assert len(merged.events) == 2
assert merge_records(merged, left).to_dict() == merged.to_dict()
print("winner for scalar title:", merged.title)

winner for scalar title: desktop

Both append-only messages survive, while the newer scalar title wins deterministically. Determinism is not the same as good conflict experience: a hosted UI should surface meaningful scalar conflicts instead of quietly hiding the losing value. The merge algebra guarantees convergence; product policy decides what users can inspect and undo.

Token lifecycle includes revocation and failure states

A signed token proves that the server issued claims that have not expired or been revoked; it does not prove the current request may access every session. Real authorization also checks ownership or membership after authentication. Keep the signing secret in an environment injector or managed secret store, never in browser JavaScript, a notebook, a compose file, or a recorded trajectory.

from autocode_service.auth import TokenAuthority

authority = TokenAuthority("ephemeral-test-secret")
token = authority.issue("desktop", ttl=60)
claims = authority.verify(token)
assert claims["device_id"] == "desktop"

authority.revoke(token)
try:
    authority.verify(token)
except PermissionError as error:
    result = str(error)
else:
    raise AssertionError("revoked token was accepted")

assert result == "invalid or revoked token"
print("revocation response:", result)

revocation response: invalid or revoked token

The route translates this domain rejection to HTTP 401 and includes WWW-Authenticate: Bearer. Expired or revoked credentials are not retried automatically with the same token. Idempotent data writes may be retried after obtaining fresh credentials because the idempotency key identifies the logical push independently of authentication.

Exercises

Add session-level authorization after token verification. Specify the ownership or membership lookup, status codes that avoid leaking private session existence, test cases for revoked and cross-session tokens, and the relationship between authorization and idempotent retry.

[P07.1] Protect a two-writer merge

Design tests for two authorized devices that update one session concurrently. Check merge algebra, repeated idempotency keys, a revoked device, and an authenticated device that lacks permission for the target session.