From d4cda432d21ca7cca6f2c37c1236089ae326e379 Mon Sep 17 00:00:00 2001 From: Akshat malik Date: Sun, 13 Sep 2026 13:16:00 +0530 Subject: [PATCH] Refactor cookie-mode anonymous history hand-off --- examples/starter/agents.yaml | 2 +- src/agent_manager/api/deps.py | 51 ++++- .../api/static/widget/api/AgentChatClient.ts | 10 +- .../api/static/widget/auth/tokenSource.ts | 104 ++-------- tests/agent_manager/test_api.py | 186 ++++++++++++++++++ 5 files changed, 259 insertions(+), 94 deletions(-) diff --git a/examples/starter/agents.yaml b/examples/starter/agents.yaml index a1f09df7..88104b3d 100644 --- a/examples/starter/agents.yaml +++ b/examples/starter/agents.yaml @@ -7,7 +7,7 @@ system: defaults: model: provider: openai - name: qwen2.5:14b + name: openai/gpt-oss-20b temperature: 0.0 execution: diff --git a/src/agent_manager/api/deps.py b/src/agent_manager/api/deps.py index f997d4a1..93765e6e 100644 --- a/src/agent_manager/api/deps.py +++ b/src/agent_manager/api/deps.py @@ -16,6 +16,7 @@ BEARER_PREFIX = "Bearer " UNAUTHENTICATED_DETAIL = "a verified identity is required" +VISITOR_PASS_HEADER = "X-Extra-Visitor-Pass" @dataclass(frozen=True) @@ -35,7 +36,7 @@ def get_caller_identity(request: Request) -> CallerIdentity: return request.app.state.caller_identity -def get_principal(request: Request) -> Principal: +async def get_principal(request: Request) -> Principal: """The proven caller every conversation route authorizes against. A bearer token where the caller supplied one, otherwise the host's session @@ -43,17 +44,63 @@ def get_principal(request: Request) -> Principal: host's own origin: cross-site requests cannot read a JSON response, and the widget's `application/json` writes are preflighted against a CORS allowlist that denies by default. + + After resolving an authenticated (non-anonymous) principal, the dependency + opportunistically adopts any anonymous history the widget attached via the + X-Extra-Visitor-Pass header. Failures are logged and silently swallowed so + an adoption hiccup never blocks the actual request. """ identity = get_caller_identity(request) token = _select_token(request, identity) if token is None: raise HTTPException(status_code=401, detail=UNAUTHENTICATED_DETAIL) try: - return identity.resolver.resolve(token) + principal = identity.resolver.resolve(token) except TokenError as exc: logger.warning("token verification failed: %s", exc) raise HTTPException(status_code=401, detail=str(exc)) from None + # Server-side opportunistic hand-off. + # When both an authenticated principal AND a visitor pass arrive on the same + # request, adopt the anonymous history before the route runs. This removes + # the need for the frontend to repeatedly probe /auth/link in cookie mode. + if not principal.is_anonymous: + raw_pass = request.headers.get(VISITOR_PASS_HEADER) + if raw_pass: + await _try_adopt_visitor_history(request, raw_pass, principal, identity) + + return principal + + +async def _try_adopt_visitor_history( + request: Request, + raw_pass: str, + principal: Principal, + identity: CallerIdentity, +) -> None: + """Opportunistically adopt anonymous history. Failures never block the request. + + Invalid/expired pass: TokenError is caught and logged at DEBUG — the + authenticated request continues normally and the stale pass is effectively + discarded. + + Temporary DB failure: Exception is caught and logged at WARNING — the pass + is NOT marked consumed so the next request will retry automatically. + + Already-adopted pass: link_anonymous_user runs an atomic UPDATE WHERE + linked_to_user_id IS NULL, which affects 0 rows and returns cleanly. + """ + try: + visitor = identity.resolver.anonymous.resolve(raw_pass) + except TokenError: + logger.debug("X-Extra-Visitor-Pass is invalid or expired; ignoring") + return + try: + service = get_service(request) + await service.link_anonymous(visitor, principal) + except Exception: + logger.warning("opportunistic anonymous history adoption failed", exc_info=True) + Service = Annotated[ConversationService, Depends(get_service)] Caller = Annotated[Principal, Depends(get_principal)] diff --git a/src/agent_manager/api/static/widget/api/AgentChatClient.ts b/src/agent_manager/api/static/widget/api/AgentChatClient.ts index 504d86b8..a6df3d8e 100644 --- a/src/agent_manager/api/static/widget/api/AgentChatClient.ts +++ b/src/agent_manager/api/static/widget/api/AgentChatClient.ts @@ -43,8 +43,8 @@ export class AgentChatClient { /** A 401 usually means the token expired: renew once and retry. The rejected * attempt changed nothing, so replaying is safe. */ - private async request(path: string, init?: RequestInit, options?: { forceCheck?: boolean }): Promise { - let response = await this.send(path, init, await this.tokens.current({ forceCheck: options?.forceCheck })); + private async request(path: string, init?: RequestInit): Promise { + let response = await this.send(path, init, await this.tokens.current()); if (response.status === 401) { response = await this.send(path, init, await this.tokens.renew()); } @@ -57,6 +57,10 @@ export class AgentChatClient { private send(path: string, init: RequestInit | undefined, token: string | null) { const headers: Record = { "Content-Type": "application/json" }; if (token) headers.Authorization = `Bearer ${token}`; + // Carry the visitor pass on every request so the server can opportunistically + // adopt anonymous history the moment it sees an authenticated principal. + const pass = this.tokens.visitorPass; + if (pass) headers["X-Extra-Visitor-Pass"] = pass; // `include` lets a same-origin deployment authenticate by the host's own // cookie, which the widget can never read. return fetch(`${this.endpoint}${path}`, { ...init, headers, credentials: "include" }); @@ -71,7 +75,7 @@ export class AgentChatClient { async listConversations(limit = 20, cursor?: string | null): Promise { const params = new URLSearchParams({ limit: String(limit) }); if (cursor) params.set("cursor", cursor); - const response = await this.request(`/conversations?${params.toString()}`, undefined, { forceCheck: true }); + const response = await this.request(`/conversations?${params.toString()}`); const data = await response.json(); const rawItems: Array<{ conversation_id: string; title?: string | null; last_message_at?: string | null }> = diff --git a/src/agent_manager/api/static/widget/auth/tokenSource.ts b/src/agent_manager/api/static/widget/auth/tokenSource.ts index 71a6c918..8b84a126 100644 --- a/src/agent_manager/api/static/widget/auth/tokenSource.ts +++ b/src/agent_manager/api/static/widget/auth/tokenSource.ts @@ -34,7 +34,6 @@ export interface TokenSourceOptions { } const PASS_ENDPOINT = "/auth/anonymous"; -const LINK_ENDPOINT = "/auth/link"; export function visitorPassKey(endpoint: string): string { return `agent-chat:pass:${endpoint}`; @@ -46,10 +45,6 @@ export class TokenSource { /** Bumped by `reset()` so a resolution already in flight, once it lands, can * tell it is answering a question nobody is asking anymore. */ private generation = 0; - /** Avoid repeating unauthenticated claim attempts for the same pass in cookie mode. */ - private lastClaimAttemptPass: string | null = null; - private lastCookieSnapshot = typeof document !== "undefined" ? document.cookie : ""; - private identityCheckDirty = true; private readonly tokenUrl: string; private readonly provider: TokenProvider | null; private readonly storage: Storage; @@ -65,43 +60,28 @@ export class TokenSource { this.storage = options.storage ?? localStorage; this.requireIdentity = options.requireIdentity ?? false; this.onIdentityFailure = options.onIdentityFailure ?? (() => {}); - - if (typeof window !== "undefined") { - const markDirty = () => { - this.identityCheckDirty = true; - }; - try { - window.addEventListener("focus", markDirty); - if (typeof document !== "undefined") { - document.addEventListener("visibilitychange", markDirty); - } - window.addEventListener("storage", markDirty); - } catch { - // Non-browser or custom environment ignore - } - } } - async current(options?: { forceCheck?: boolean }): Promise { - const pass = this.storedPass(); - const cookieChanged = this.cookieSnapshotChanged(); - const force = options?.forceCheck ?? false; - const shouldClaim = - this.isCookieMode() && - pass !== null && - (force || this.identityCheckDirty || cookieChanged || pass !== this.lastClaimAttemptPass); + /** The stored anonymous visitor pass, if one exists. + * + * Sent on every request as `X-Extra-Visitor-Pass` so the server can + * opportunistically adopt anonymous history the moment it sees an + * authenticated principal alongside it. The server validates and consumes it; + * the frontend just carries it passively until cleared post-adoption. + */ + get visitorPass(): string | null { + return this.storedPass(); + } - if (!this.cached || shouldClaim) { + async current(): Promise { + if (!this.cached) { await this.resolve(() => this.storedPass()); - this.identityCheckDirty = false; - this.updateCookieSnapshot(); } return this.cached; } /** After a 401: whatever we sent is no good, so get another. */ async renew(): Promise { - this.identityCheckDirty = true; return this.resolve(() => this.issuePass()); } @@ -116,9 +96,6 @@ export class TokenSource { // next call starts a fresh one instead of awaiting an answer to a question // that no longer applies (e.g. the old tokenProvider). this.pending = null; - this.lastClaimAttemptPass = null; - this.identityCheckDirty = true; - this.updateCookieSnapshot(); } /** Drop this browser's identity entirely — a host app signing its user out. */ @@ -127,17 +104,6 @@ export class TokenSource { this.clearPass(); } - private cookieSnapshotChanged(): boolean { - if (typeof document === "undefined") return false; - return document.cookie !== this.lastCookieSnapshot; - } - - private updateCookieSnapshot(): void { - if (typeof document !== "undefined") { - this.lastCookieSnapshot = document.cookie; - } - } - /** Concurrent callers share one resolution. Without this, parallel requests * each fetch a token and each hand over the visitor pass. */ private resolve(fallback: () => string | null | Promise): Promise { @@ -160,49 +126,11 @@ export class TokenSource { return this.pending; } - private isCookieMode(): boolean { - return !this.tokenUrl && !this.provider; - } - - /** A host token, plus the one-time hand-off of whatever this browser chatted - * about before signing in. */ + /** A host token only. In cookie mode this returns null — the session cookie + * speaks for the caller directly. Anonymous history adoption is now handled + * server-side via the X-Extra-Visitor-Pass header. */ private async hostToken(): Promise { - const token = await this.fromHost(); - if (token || (this.isCookieMode() && this.storedPass())) { - await this.claimVisitorHistory(token); - } - return token; - } - - private async claimVisitorHistory(hostToken: string | null): Promise { - const pass = this.storedPass(); - if (!pass) return; - try { - const headers: Record = { "Content-Type": "application/json" }; - if (hostToken) headers.Authorization = `Bearer ${hostToken}`; - const response = await fetch(`${this.endpoint}${LINK_ENDPOINT}`, { - method: "POST", - headers, - credentials: "include", - body: JSON.stringify({ anonymous_token: pass }), - }); - if (response.ok) { - const data = (await response.json().catch(() => null)) as { conversations_moved?: number } | null; - if (hostToken !== null || (data?.conversations_moved ?? 0) > 0) { - this.clearPass(); - this.lastClaimAttemptPass = null; - this.identityCheckDirty = true; - } else { - this.lastClaimAttemptPass = pass; - } - } else if (response.status === 401) { - this.lastClaimAttemptPass = pass; - } else if (hostToken !== null && response.status >= 400 && response.status < 500) { - this.clearPass(); - } - } catch { - // Offline: keep the pass so the next page load retries the hand-off. - } + return this.fromHost(); } private clearPass(): void { diff --git a/tests/agent_manager/test_api.py b/tests/agent_manager/test_api.py index 408abb58..173b1744 100644 --- a/tests/agent_manager/test_api.py +++ b/tests/agent_manager/test_api.py @@ -1434,3 +1434,189 @@ def test_tool_error_text_is_sanitized_in_stream_message() -> None: assert response.status_code == 200 assert "Tool execution failed" in response.text assert "localhost" not in response.text + + +# ── Anonymous history hand-off regression tests ────────────────────────────── +# Covers the server-side opportunistic hand-off introduced to replace the +# repeated POST /auth/link → 401 probing pattern. + +_VISITOR_PASS_HEADER = "X-Extra-Visitor-Pass" + + +def _cookie_app(): + """A host-token (cookie-mode) app reused by several hand-off tests.""" + return build_test_app( + ConversationService(RecordingEngine(), MemoryRepository()), + extra_auth_mode=AuthMode.HOST_TOKEN, + extra_auth_cookie=HOST_COOKIE, + extra_auth_claim_user_id="id", + ) + + +def test_repeated_anonymous_list_conversations_never_probes_auth_link() -> None: + """While still anonymous, repeated GET /conversations must work cleanly with + no /auth/link probe required. + + The old implementation issued POST /auth/link → 401 on every forced + identity check. The new server-side hand-off has no such probe — the + anonymous request simply returns its own (empty) list. + """ + app = _cookie_app() + client = TestClient(app) + pass_token = client.post("/auth/anonymous").json()["token"] + visitor = {"Authorization": f"Bearer {pass_token}"} + + for _ in range(3): + resp = client.get("/conversations", headers=visitor) + assert resp.status_code == 200 + assert resp.json()["items"] == [] + + +def test_first_authenticated_request_adopts_anonymous_history_via_header() -> None: + """Server-side opportunistic hand-off: the first GET /conversations that + arrives with both an authenticated cookie and X-Extra-Visitor-Pass merges + the anonymous history before returning — no reset() or refreshIdentity(). + """ + app = _cookie_app() + anon_client = TestClient(app) + pass_token = anon_client.post("/auth/anonymous").json()["token"] + visitor = {"Authorization": f"Bearer {pass_token}"} + + cid = anon_client.post("/conversations", headers=visitor).json()["conversation_id"] + anon_client.post( + f"/conversations/{cid}/messages", json={"message": "pre-login"}, headers=visitor + ) + + # Host app sets session cookie (user signs in). Widget sends pass in new header. + alice = TestClient(app, cookies=session_cookie(id="alice")) + resp = alice.get("/conversations", headers={_VISITOR_PASS_HEADER: pass_token}) + + assert resp.status_code == 200 + conv_ids = [t["conversation_id"] for t in resp.json()["items"]] + assert cid in conv_ids # merged on the very first request — no delay + assert alice.get(f"/conversations/{cid}/messages").status_code == 200 + + +def test_pagination_after_adoption_does_not_re_adopt() -> None: + """Paging through history after the hand-off must not cause repeated + adoptions. The already-adopted pass returns 0 rows moved — idempotent. + """ + app = _cookie_app() + anon_client = TestClient(app) + pass_token = anon_client.post("/auth/anonymous").json()["token"] + visitor = {"Authorization": f"Bearer {pass_token}"} + + for i in range(3): + cid = anon_client.post("/conversations", headers=visitor).json()["conversation_id"] + anon_client.post( + f"/conversations/{cid}/messages", json={"message": f"msg {i}"}, headers=visitor + ) + + alice = TestClient(app, cookies=session_cookie(id="alice")) + hand_off = {_VISITOR_PASS_HEADER: pass_token} + + page1 = alice.get("/conversations?limit=2", headers=hand_off) + assert page1.status_code == 200 + cursor = page1.json().get("next_cursor") + + if cursor: + page2 = alice.get(f"/conversations?limit=2&cursor={cursor}", headers=hand_off) + assert page2.status_code == 200 + + +def test_already_adopted_pass_in_header_is_a_no_op() -> None: + """A pass that was already consumed moves 0 rows and does not error. + The request continues normally and conversations are not duplicated. + """ + app = build_test_app(ConversationService(RecordingEngine(), MemoryRepository())) + client = TestClient(app) + pass_token = client.post("/auth/anonymous").json()["token"] + visitor = {"Authorization": f"Bearer {pass_token}"} + + client.post("/conversations", headers=visitor) + alice = bearer("alice") + + r1 = client.get("/conversations", headers={**alice, _VISITOR_PASS_HEADER: pass_token}) + assert r1.status_code == 200 + assert len(r1.json()["items"]) == 1 + + # Second request with same pass — adoption already done, must be a no-op. + r2 = client.get("/conversations", headers={**alice, _VISITOR_PASS_HEADER: pass_token}) + assert r2.status_code == 200 + assert len(r2.json()["items"]) == 1 # still exactly 1 — not duplicated + + +def test_invalid_visitor_pass_in_header_does_not_block_request() -> None: + """An expired or malformed pass in X-Extra-Visitor-Pass must never turn a + valid authenticated request into a 4xx. The pass is silently discarded. + """ + app = build_test_app(ConversationService(RecordingEngine(), MemoryRepository())) + client = TestClient(app) + alice = bearer("alice") + + resp = client.get("/conversations", headers={**alice, _VISITOR_PASS_HEADER: "not-a-valid-jwt"}) + + assert resp.status_code == 200 # request succeeds despite the bad pass + assert resp.json()["items"] == [] + + +def test_concurrent_authenticated_requests_adopt_history_exactly_once() -> None: + """Two requests arriving post-login must result in exactly one adoption. + The SQL atomic UPDATE (WHERE linked_to_user_id IS NULL) guarantees this. + """ + app = build_test_app(ConversationService(RecordingEngine(), MemoryRepository())) + client = TestClient(app) + pass_token = client.post("/auth/anonymous").json()["token"] + visitor = {"Authorization": f"Bearer {pass_token}"} + + cid = client.post("/conversations", headers=visitor).json()["conversation_id"] + alice = bearer("alice") + hand_off = {**alice, _VISITOR_PASS_HEADER: pass_token} + + # Simulate two "concurrent" requests sequentially — the atomic SQL guard + # makes the second a no-op regardless of timing. + r1 = client.get("/conversations", headers=hand_off) + r2 = client.get("/conversations", headers=hand_off) + + assert r1.status_code == 200 + assert r2.status_code == 200 + all_ids = [t["conversation_id"] for t in r2.json()["items"]] + assert all_ids.count(cid) == 1 # not duplicated + + +def test_bearer_mode_auth_link_still_works_after_refactor() -> None: + """The /auth/link endpoint must continue to work for explicit bearer/token + mode. This is a direct regression guard against the old link path. + """ + app = build_test_app(ConversationService(RecordingEngine(), MemoryRepository())) + client = TestClient(app) + pass_token = client.post("/auth/anonymous").json()["token"] + visitor = {"Authorization": f"Bearer {pass_token}"} + + cid = client.post("/conversations", headers=visitor).json()["conversation_id"] + alice = bearer("alice") + + linked = client.post("/auth/link", json={"anonymous_token": pass_token}, headers=alice) + + assert linked.status_code == 200 + assert linked.json()["conversations_moved"] == 1 + assert [ + t["conversation_id"] for t in client.get("/conversations", headers=alice).json()["items"] + ] == [cid] + + +def test_anonymous_request_without_pass_header_is_unaffected() -> None: + """A plain anonymous request (no X-Extra-Visitor-Pass, no cookie) must + still work normally — the hand-off header is purely additive. + """ + app = build_test_app(ConversationService(RecordingEngine(), MemoryRepository())) + client = TestClient(app) + pass_token = client.post("/auth/anonymous").json()["token"] + visitor = {"Authorization": f"Bearer {pass_token}"} + + cid = client.post("/conversations", headers=visitor).json()["conversation_id"] + client.post(f"/conversations/{cid}/messages", json={"message": "hi"}, headers=visitor) + + resp = client.get("/conversations", headers=visitor) + assert resp.status_code == 200 + assert resp.json()["items"][0]["conversation_id"] == cid