The agent will now use your own tools when you ask it something in " f"Slack, as {html.escape(email or 'your SGP account')}.
" - f"This connection is valid until {html.escape(when)}, after " - "which the agent will ask you to reconnect. You can close this page and go " - "back to Slack.
", + + ( + "I'm answering the message you sent in Slack now — head back to that " + "thread.
" + if replayed + else "Go back to Slack and ask again.
" + ) + + f"This connection is valid until {html.escape(when)}, after " + "which the agent will ask you to reconnect.
", ) diff --git a/agentex/src/domain/use_cases/slack_gateway_use_case.py b/agentex/src/domain/use_cases/slack_gateway_use_case.py index b7f938a3..cac4b5eb 100644 --- a/agentex/src/domain/use_cases/slack_gateway_use_case.py +++ b/agentex/src/domain/use_cases/slack_gateway_use_case.py @@ -251,7 +251,11 @@ def _strip_selector(text: str, selector: str) -> str: def _turn_content( - inbound: InboundSlack, prompt: str, *, self_posts: bool = False + inbound: InboundSlack, + prompt: str, + *, + self_posts: bool = False, + unlinked: bool = False, ) -> str: """Prepend the Slack conversation context to the turn. @@ -263,12 +267,37 @@ def _turn_content( ``self_posts`` is set for agents the gateway does NOT relay (golden-agent, which has Slack write tools). For them we add an explicit directive to post their own reply into this thread, because nothing is delivered on their behalf — without it the turn - would run and produce text that never reaches Slack.""" + would run and produce text that never reaches Slack. + + ``unlinked`` tells the agent it is NOT running as the person who asked. Without it + the turn runs on the shared bot identity and cannot tell — so it answers about + *its own* access as though it were theirs. Observed: confidently reporting "no + Linear access, verified two ways" and separately overstating GitHub, while the + gateway was simultaneously DMing that person a link to connect. Two messages that + contradict each other, one of them wrong about the user's capabilities. + + It also stops the agent inventing its own authorization flow. Lacking a + credential, the harness offers an OAuth URL with a localhost redirect — a dead end + from Slack, and indistinguishable from a real instruction. Telling the agent a + link has already been sent removes the reason to improvise one.""" context = ( f"[Slack context] channel_id={inbound.channel} thread_ts={inbound.thread_ts}. " f"This message came from that Slack thread; to read earlier messages or the " f"channel's history, use your Slack tools with this channel_id." ) + if unlinked: + context += ( + " IMPORTANT: you are NOT running as the person who sent this message. This " + "turn uses a shared service identity, so you do NOT have access to their " + "personal integrations (Notion, Linear, GitHub, …) — any such tool you can " + "reach belongs to the shared account, not to them. Do not describe your own " + "access as if it were theirs, and do not conclude anything about what they " + "have connected. They have ALREADY been sent a link to connect their " + "account, so do NOT generate authorization or OAuth links of your own. " + "Answer what you can from this thread and from shared tools; if the request " + "needs their personal data, say plainly that it needs their account " + "connected first and that a link has been sent." + ) if self_posts: context += ( " IMPORTANT: your text response is NOT posted to Slack for you. Deliver your " @@ -652,7 +681,9 @@ async def _submit_agents_modal( background.add_task(self._run_turn, inbound) return {} # empty 200 closes the modal - async def _run_turn(self, inbound: InboundSlack) -> None: + async def _run_turn( + self, inbound: InboundSlack, *, offer_link: bool = True + ) -> None: try: # Whose identity runs this turn. # @@ -668,12 +699,18 @@ async def _run_turn(self, inbound: InboundSlack) -> None: # personal integrations. Prompting them to link is a separate concern. principal, auth_headers, sgp_user_id = await self._turn_identity(inbound) - if sgp_user_id is None: + if sgp_user_id is None and offer_link: # Not running as a person: either unlinked, or linked with a # credential we can't use (expired session, undecryptable). Offer the # link either way — a dead credential needs the same fix as no # credential. Rate-limited and best-effort; it never affects the turn, # which continues as the bot below (or is refused just after). + # + # ``offer_link`` is False for a replay fired by the link callback. That + # turn exists *because* the user just linked, so nudging them again + # would be absurd — and if the fresh link somehow still doesn't + # resolve, offering would mint another nonce, DM another link, and + # invite the same loop again on the next click. await self._offer_link(inbound) if principal is None and auth_headers is None: @@ -759,16 +796,18 @@ async def _turn_identity( identity = await self._resolve_invoking_identity(inbound) if identity is not None: headers = await self._identity_link_service().acting_headers(identity) - if headers is not None: + if headers is not None and await self._credential_still_accepted( + identity, headers + ): logger.info( "[slack] turn acting as sgp user %s (linked from %s)", identity.sgp_user_id, inbound.user, ) return identity.principal, headers, identity.sgp_user_id - # Linked but unusable — no stored key, expired, or undecryptable. The - # service has already logged which. Treated the same as unlinked here; - # re-link prompting is handled separately. + # Linked but unusable — no stored credential, locally expired, + # undecryptable, or rejected upstream. Falls through to the bot with + # sgp_user_id=None, which is what makes the caller offer a re-link. if _REQUIRE_LINKED_USER: logger.info( @@ -785,6 +824,134 @@ async def _turn_identity( ) return bot_principal, bot_headers, None + async def replay_pending_turn( + self, *, team_id: str, user_id: str, pending_turn: dict[str, Any] | None + ) -> bool: + """Re-run the message that prompted a link, now that the user has linked. + + Called by the link callback. Without it the flow is ask -> get a link -> + click -> **ask again**, and the question the person actually had is dropped on + the floor at the exact moment we finally could have answered it. + + The replay lands in a *new* task, for free: the task key includes the SGP user + id, so the same Slack thread keys differently once linked + (``slack:{ts}`` -> ``slack:{team}:{channel}:{ts}:{sgp_user}``). That means a + fresh turn-1 session with their credentials — none of the toolset pinning that + makes enabling an MCP mid-conversation a no-op. The pre-link exchange isn't in + that task's history, but the agent can read the Slack thread with its own + tools, which the context prefix tells it how to do. + + Returns whether a replay was started. Best-effort by contract: the caller has + already stored the link and must not fail, or roll it back, because a replay + didn't happen. + """ + if not pending_turn: + return False + text = (pending_turn.get("text") or "").strip() + channel = pending_turn.get("channel") or "" + if not (text and channel and team_id and user_id): + logger.info( + "[slack] link replay skipped: incomplete pending turn", + extra={"has_text": bool(text), "has_channel": bool(channel)}, + ) + return False + + inbound = InboundSlack( + team_id=team_id, + channel=channel, + user=user_id, + text=text, + thread_ts=pending_turn.get("thread_ts") or "", + # Re-derived exactly as normalize() does, so a selector-driven turn + # ("@agent some-config ...") resolves to the same target it would have. + selector=text.split(maxsplit=1)[0] if text else None, + ) + # The indicator matters here: the answer arrives minutes after the user + # clicked a web page, so without it a reply appears from nowhere. + await self._set_status(inbound, "is thinking…") + # offer_link=False: see _run_turn. A replay must never nudge again. + await self._run_turn(inbound, offer_link=False) + return True + + async def _credential_still_accepted( + self, identity: Any, headers: dict[str, str] + ) -> bool: + """Whether the stored session is still accepted upstream. + + ``credential_expires_at`` is an upper bound, not a guarantee. A session JWT + stops being honoured the moment its owner signs out or it is revoked, while + the stored expiry still reads months away — so without this check the gateway + hands the agent a dead cookie, every user-scoped tool call fails with a 401 + the agent can only report as "I can't reach your Notion", and nothing ever + prompts a re-link because locally the credential looks fine. + + **A rejection is not the same as a failure**, and conflating them is the trap + here. If the auth service is misconfigured or down it will reject or error on + *everyone*, and concluding "your credential is bad" would tell an entire + workspace to re-link — which would fix nothing, because nothing is wrong with + their credentials. The distinction comes from the adapter's own exception + types rather than from guesswork: + + - ``AuthenticationError`` / ``AuthorizationError`` — this credential was + refused. Re-linking is the remedy, so report it unusable. 403 counts: a + valid session that can no longer use the stored account needs a re-link to + pick up a current one. + - anything else (gateway error, service unavailable, timeout, surprise) — + the *check* failed, not the credential. Assume it's good and proceed. An + outage in a verification step must not revoke everyone's access. + + Deliberately non-destructive: nothing is deleted or tombstoned on rejection. + The row keeps its credential, so a systemic 401 costs a bot-fallback turn and + a rate-limited nudge, and recovers by itself when the upstream does. Cheap + enough to run per turn — the shared-bot path already verifies its own key on + every turn, so this is the same cost profile, not a new one. + """ + # Local imports avoid an import cycle at module load. + from src.adapters.authentication.adapter_agentex_authn_proxy import ( + AgentexAuthenticationProxy, + ) + from src.config.dependencies import resolve_environment_variable_dependency + from src.config.environment_variables import EnvVarKeys + + # ClientError is the 4xx base and a *sibling* of ServiceError, not a parent — + # so catching it gets "your credential was refused" and never "the service + # broke". That split is the whole reason this can be safe. + from src.domain.exceptions import ClientError + + try: + auth_url = resolve_environment_variable_dependency( + EnvVarKeys.AGENTEX_AUTH_URL + ) + except Exception: # noqa: BLE001 - authz off locally: nothing to verify against + return True + if not auth_url: + return True + + authn = AgentexAuthenticationProxy( + agentex_auth_url=auth_url, + environment=resolve_environment_variable_dependency(EnvVarKeys.ENVIRONMENT), + ) + try: + await authn.verify_headers(dict(headers)) + return True + except ClientError as exc: + # 4xx: upstream looked at this credential and refused it. + logger.info( + "identity_link_credential_rejected", + extra={ + "sgp_user_id": getattr(identity, "sgp_user_id", None), + "reason": type(exc).__name__, + }, + ) + return False + except Exception: # noqa: BLE001 - the check failed, not the credential + logger.warning( + "identity_link_credential_check_failed", + extra={"sgp_user_id": getattr(identity, "sgp_user_id", None)}, + exc_info=True, + ) + return True + def _identity_link_service(self): """Build the identity-link service. @@ -988,7 +1155,14 @@ async def _dispatch( content = TextContentEntity( author=MessageAuthor.USER, content=_turn_content( - inbound, prompt, self_posts=target.agent_name == _DEFAULT_AGENT_NAME + inbound, + prompt, + self_posts=target.agent_name == _DEFAULT_AGENT_NAME, + # sgp_user_id is None exactly when the turn is NOT running as a + # person — the same signal that triggers the link offer. So the agent + # is told it's unlinked precisely when the user is being asked to + # link, and the two messages agree instead of contradicting. + unlinked=sgp_user_id is None, ), format=TextFormat.MARKDOWN, ) diff --git a/agentex/tests/unit/api/test_integrations_routes.py b/agentex/tests/unit/api/test_integrations_routes.py index c31ea3eb..9c229c23 100644 --- a/agentex/tests/unit/api/test_integrations_routes.py +++ b/agentex/tests/unit/api/test_integrations_routes.py @@ -17,6 +17,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from fastapi import BackgroundTasks from src.api.routes import integrations as mod from src.domain.entities.identity_links import IdentityLinkMethod, IdentityProvider from src.domain.services.link_nonce_service import LinkRequest @@ -73,6 +74,15 @@ def _link_request(**kw) -> LinkRequest: } +async def _confirm(request, nonce: str = "tok", background=None): + """Call the confirm handler with a BackgroundTasks, which it needs for the replay.""" + return await mod.slack_link_confirm( + request, + background if background is not None else BackgroundTasks(), + nonce=nonce, + ) + + @pytest.fixture def wiring(monkeypatch): """Stub the two remaining collaborators: nonce store and repository.""" @@ -133,7 +143,7 @@ async def test_display_name_is_html_escaped(self, wiring): @pytest.mark.asyncio class TestConfirm: async def test_stores_the_callers_session_and_invalidates(self, wiring): - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 200 assert "connected" in resp.body.decode().lower() @@ -154,7 +164,7 @@ async def test_stores_the_callers_session_and_invalidates(self, wiring): wiring.service.invalidate.assert_awaited_once() async def test_expiry_comes_from_the_token_not_a_guessed_ttl(self, wiring): - await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + await _confirm(_request(_PRINCIPAL), nonce="tok") stored = wiring.repo.upsert_link.await_args.kwargs["credential_expires_at"] # ~150 days out, from the JWT's own exp claim. assert 149 <= (stored - datetime.now(UTC)).days <= 150 @@ -163,7 +173,7 @@ async def test_token_without_an_exp_gets_a_bounded_fallback(self, wiring): # Never store an unbounded credential: an unknown expiry becomes a short # known one rather than "valid forever". req = _request(_PRINCIPAL, cookie=f"_identityJwt={_jwt(None)}") - resp = await mod.slack_link_confirm(req, nonce="tok") + resp = await _confirm(req, nonce="tok") assert resp.status_code == 200 stored = wiring.repo.upsert_link.await_args.kwargs["credential_expires_at"] @@ -175,7 +185,7 @@ async def test_already_expired_token_is_refused(self, wiring): _PRINCIPAL, cookie=f"_identityJwt={_jwt(datetime.now(UTC) - timedelta(minutes=1))}", ) - resp = await mod.slack_link_confirm(req, nonce="tok") + resp = await _confirm(req, nonce="tok") assert resp.status_code == 401 # Storing it would create a link that can never work. @@ -186,7 +196,7 @@ async def test_no_session_cookie_stores_nothing(self, wiring): # Authenticated by an api-key or bearer rather than a browser session: there # is nothing here we could act through later. req = _request(_PRINCIPAL, cookie="_ga=GA1.2.x; csrftoken=abc") - resp = await mod.slack_link_confirm(req, nonce="tok") + resp = await _confirm(req, nonce="tok") assert resp.status_code == 400 wiring.repo.upsert_link.assert_not_awaited() @@ -196,24 +206,24 @@ async def test_principal_without_an_account_is_refused(self, wiring): # The secrets service requires account context, so a link without one would # look connected and resolve nothing. req = _request({**_PRINCIPAL, "account_id": None}) - resp = await mod.slack_link_confirm(req, nonce="tok") + resp = await _confirm(req, nonce="tok") assert resp.status_code == 400 assert "account" in resp.body.decode().lower() wiring.repo.upsert_link.assert_not_awaited() async def test_nonce_is_consumed_only_after_a_successful_store(self, wiring): - await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + await _confirm(_request(_PRINCIPAL), nonce="tok") wiring.nonce.consume.assert_awaited_once() async def test_expired_nonce_is_refused(self, wiring): wiring.nonce.peek = AsyncMock(return_value=None) - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="stale") + resp = await _confirm(_request(_PRINCIPAL), nonce="stale") assert resp.status_code == 400 wiring.repo.upsert_link.assert_not_awaited() async def test_unauthenticated_stores_nothing(self, wiring): - resp = await mod.slack_link_confirm(_request(None), nonce="tok") + resp = await _confirm(_request(None), nonce="tok") assert resp.status_code == 401 wiring.repo.upsert_link.assert_not_awaited() @@ -223,7 +233,7 @@ async def test_sgp_account_already_linked_to_another_slack_user(self, wiring): wiring.repo.get_active_by_sgp_user = AsyncMock( return_value=SimpleNamespace(external_user_id="U-someone-else") ) - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 409 assert "already linked" in resp.body.decode().lower() @@ -235,7 +245,7 @@ async def test_relinking_the_same_slack_user_is_allowed(self, wiring): wiring.repo.get_active_by_sgp_user = AsyncMock( return_value=SimpleNamespace(external_user_id="U1") ) - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 200 wiring.repo.upsert_link.assert_awaited_once() @@ -247,7 +257,7 @@ async def test_unconfigured_encryption_key_is_reported_not_a_500(self, wiring): wiring.repo.upsert_link = AsyncMock( side_effect=mod.CredentialEncryptionError("key unset") ) - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 503 assert ( @@ -289,19 +299,19 @@ def _slack_profile(self, monkeypatch, **profile): async def test_matching_emails_link_successfully(self, wiring, monkeypatch): self._slack_profile(monkeypatch, email=_SGP_EMAIL) - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 200 wiring.repo.upsert_link.assert_awaited_once() async def test_match_ignores_case_and_surrounding_space(self, wiring, monkeypatch): self._slack_profile(monkeypatch, email=f" {_SGP_EMAIL.upper()} ") - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 200 async def test_mismatch_is_refused_and_stores_nothing(self, wiring, monkeypatch): self._slack_profile(monkeypatch, email="attacker@example.com") - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 403 wiring.repo.upsert_link.assert_not_awaited() @@ -316,7 +326,7 @@ async def test_missing_scope_allows_the_link(self, wiring, monkeypatch): # won't tell us. Refusing here would block every link in the workspace, so # the check stands down rather than failing closed. self._slack_profile(monkeypatch, email=None, error="missing_scope") - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 200 wiring.repo.upsert_link.assert_awaited_once() @@ -325,23 +335,21 @@ async def test_lookup_failure_allows_the_link(self, wiring, monkeypatch): # would make linking fail randomly, and the gap isn't attacker-reachable: # nobody outside our infrastructure decides whether our Slack call succeeds. self._slack_profile(monkeypatch, email=None, error="request_failed") - resp = await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + resp = await _confirm(_request(_PRINCIPAL), nonce="tok") assert resp.status_code == 200 async def test_missing_sgp_email_allows_the_link(self, wiring, monkeypatch): # Slack told us, but the SGP principal carries no email: nothing to compare # against, so the same "can't verify, don't block" rule applies. self._slack_profile(monkeypatch, email="someone@example.com") - resp = await mod.slack_link_confirm( - _request({**_PRINCIPAL, "raw_user": {}}), nonce="tok" - ) + resp = await _confirm(_request({**_PRINCIPAL, "raw_user": {}}), nonce="tok") assert resp.status_code == 200 async def test_the_check_runs_on_every_confirm(self, wiring, monkeypatch): # No flag means no way to accidentally leave it off: granting the scope is # the only thing standing between here and enforcement. probe = self._slack_profile(monkeypatch, email=_SGP_EMAIL) - await mod.slack_link_confirm(_request(_PRINCIPAL), nonce="tok") + await _confirm(_request(_PRINCIPAL), nonce="tok") probe.assert_awaited_once() assert probe.await_args.args[0] == "U1" @@ -422,9 +430,7 @@ async def test_unnamed_identity_says_the_check_is_unavailable( async def test_success_page_does_not_print_the_uuid(self, wiring, monkeypatch): self._no_slack_lookup(monkeypatch) - resp = await mod.slack_link_confirm( - _request({**_PRINCIPAL, "raw_user": {}}), nonce="tok" - ) + resp = await _confirm(_request({**_PRINCIPAL, "raw_user": {}}), nonce="tok") assert resp.status_code == 200 assert _SGP_USER not in resp.body.decode() @@ -492,3 +498,66 @@ def test_emitted_name_stays_canonical(self, monkeypatch): self._allow(monkeypatch, "_identityJwt,_jwt") assert session_cookie_names() == ("_identityJwt", "_jwt") assert session_cookie_name() == "_identityJwt" + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestPendingTurnReplay: + """Linking should answer the question that prompted it. + + Without a replay the flow is ask -> get a link -> click -> ASK AGAIN, dropping the + person's actual question at the exact moment we could finally answer it. + """ + + async def test_replay_is_scheduled_after_a_successful_link(self, wiring): + bg = BackgroundTasks() + resp = await _confirm(_request(_PRINCIPAL), background=bg) + + assert resp.status_code == 200 + assert len(bg.tasks) == 1 + task = bg.tasks[0] + assert task.kwargs["team_id"] == "T1" + assert task.kwargs["user_id"] == "U1" + assert task.kwargs["pending_turn"]["text"] == "hi" + + async def test_page_says_the_answer_is_coming(self, wiring): + resp = await _confirm(_request(_PRINCIPAL)) + assert "answering the message you sent" in resp.body.decode() + + async def test_no_pending_turn_schedules_nothing(self, wiring): + wiring.nonce.peek = AsyncMock(return_value=_link_request(pending_turn=None)) + bg = BackgroundTasks() + resp = await _confirm(_request(_PRINCIPAL), background=bg) + + assert resp.status_code == 200 + assert bg.tasks == [] + # And the copy tells them to ask again, rather than promising an answer. + assert "ask again" in resp.body.decode().lower() + + async def test_a_refused_link_replays_nothing(self, wiring, monkeypatch): + # 403 on an email mismatch: no link was stored, so there is nobody to answer + # as. Replaying here would run the question as the shared bot. + import src.domain.use_cases.slack_gateway_use_case as sg + + monkeypatch.setattr( + sg, + "slack_user_profile", + AsyncMock(return_value={"email": "attacker@example.com", "error": None}), + ) + bg = BackgroundTasks() + resp = await _confirm(_request(_PRINCIPAL), background=bg) + + assert resp.status_code == 403 + assert bg.tasks == [] + wiring.repo.upsert_link.assert_not_awaited() + + async def test_replay_ordered_after_the_nonce_and_cache_work(self, wiring): + # The replay resolves the link it just created, so it must run after the cache + # invalidation — otherwise a negative cache entry makes it run as the bot. + bg = BackgroundTasks() + await _confirm(_request(_PRINCIPAL), background=bg) + + wiring.repo.upsert_link.assert_awaited_once() + wiring.nonce.consume.assert_awaited_once() + wiring.service.invalidate.assert_awaited_once() + assert len(bg.tasks) == 1 # scheduled, so it runs after this response diff --git a/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py b/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py index 1a48b5d1..4f25fd6b 100644 --- a/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py +++ b/agentex/tests/unit/use_cases/test_slack_gateway_use_case.py @@ -4,6 +4,7 @@ mocked). No running stack, no golden_agent, no Slack. """ +import contextlib import hashlib import hmac import json @@ -1916,3 +1917,396 @@ async def test_unreachable_dm_offers_nothing(self, monkeypatch): AsyncMock(return_value={"ok": False, "error": "cannot_dm_bot"}), ) assert await uc._offer_link(_inbound()) is False + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestStoredCredentialStillAccepted: + """A stored session can die before its recorded expiry — and a rejection is not + the same thing as a failure. + + credential_expires_at is an upper bound. Sign out and the JWT stops being + honoured while the stored expiry still reads months away, so without this check + the gateway hands the agent a dead cookie, every user-scoped tool call 401s, and + nothing prompts a re-link because locally the credential looks fine. + + The trap is over-applying it. A misconfigured or down auth service rejects + EVERYONE, and concluding "your credential is bad" would tell a whole workspace to + re-link — fixing nothing, since nothing is wrong with their credentials. The + adapter's exception types carry the distinction: ClientError (4xx) is a refusal + of this credential; ServiceError (5xx) and anything else mean the check broke. + """ + + def _verify(self, monkeypatch, raises=None, auth_url="http://auth.test"): + """Stub the verifier, and the env RESOLVER rather than os.environ. + + resolve_environment_variable_dependency reads a GlobalDependencies singleton + built once per process, so monkeypatch.setenv never reaches it — a test that + set the variable would silently take the "authz disabled" path and pass for + the wrong reason. + """ + import src.adapters.authentication.adapter_agentex_authn_proxy as ap + import src.config.dependencies as deps + + async def verify(_self, _headers): + if raises is not None: + raise raises + return {"user_id": "sgp-1"} + + monkeypatch.setattr(ap.AgentexAuthenticationProxy, "verify_headers", verify) + monkeypatch.setattr( + deps, + "resolve_environment_variable_dependency", + lambda key: auth_url if "AUTH_URL" in str(key) else "development", + ) + + @staticmethod + def _identity(): + return SimpleNamespace(sgp_user_id="sgp-1") + + @staticmethod + def _headers(): + return {"cookie": "_identityJwt=abc", "x-selected-account-id": "acct-1"} + + async def test_accepted_credential_is_usable(self, monkeypatch): + self._verify(monkeypatch) + uc = SlackGatewayUseCase() + assert await uc._credential_still_accepted(self._identity(), self._headers()) + + async def test_401_means_relink(self, monkeypatch): + from src.adapters.authentication.exceptions import AuthenticationError + + self._verify(monkeypatch, raises=AuthenticationError("revoked")) + uc = SlackGatewayUseCase() + assert not await uc._credential_still_accepted( + self._identity(), self._headers() + ) + + async def test_403_also_means_relink(self, monkeypatch): + # A valid session that can no longer use the stored account needs a re-link + # to capture a current one. + from src.domain.exceptions import ClientError + + class Forbidden(ClientError): + code = 403 + + self._verify(monkeypatch, raises=Forbidden("not in account")) + uc = SlackGatewayUseCase() + assert not await uc._credential_still_accepted( + self._identity(), self._headers() + ) + + @pytest.mark.parametrize( + "exc_name", + ["AuthenticationGatewayError", "AuthenticationServiceUnavailableError"], + ) + async def test_service_errors_do_not_blame_the_credential( + self, monkeypatch, exc_name + ): + # 5xx from the verifier means the CHECK failed. Revoking everyone's access + # because a verification dependency is unwell is the outage-amplifying move. + import src.adapters.authentication.exceptions as exceptions + + self._verify(monkeypatch, raises=getattr(exceptions, exc_name)("upstream")) + uc = SlackGatewayUseCase() + assert await uc._credential_still_accepted(self._identity(), self._headers()) + + @pytest.mark.parametrize("exc", [TimeoutError("slow"), RuntimeError("surprise")]) + async def test_unexpected_failures_assume_the_credential_is_fine( + self, monkeypatch, exc + ): + self._verify(monkeypatch, raises=exc) + uc = SlackGatewayUseCase() + assert await uc._credential_still_accepted(self._identity(), self._headers()) + + async def test_skipped_entirely_when_authz_is_off(self, monkeypatch): + # Local dev with no auth service: nothing to verify against, and refusing + # every credential would make the feature untestable offline. + from src.adapters.authentication.exceptions import AuthenticationError + + # Would reject if it were ever called — so a pass proves it wasn't. + self._verify(monkeypatch, raises=AuthenticationError("nope"), auth_url="") + uc = SlackGatewayUseCase() + assert await uc._credential_still_accepted(self._identity(), self._headers()) + + async def test_rejection_falls_back_to_the_bot_and_prompts(self, monkeypatch): + # End to end: a rejected credential must yield sgp_user_id=None, because that + # is precisely the condition _run_turn uses to offer a re-link. + from src.adapters.authentication.exceptions import AuthenticationError + + uc = SlackGatewayUseCase() + identity = SimpleNamespace( + sgp_user_id="sgp-1", principal={"user_id": "sgp-1"}, sgp_account_id="acct-1" + ) + monkeypatch.setattr( + uc, "_resolve_invoking_identity", AsyncMock(return_value=identity) + ) + monkeypatch.setattr( + uc, + "_identity_link_service", + lambda: SimpleNamespace( + acting_headers=AsyncMock(return_value=self._headers()) + ), + ) + self._verify(monkeypatch, raises=AuthenticationError("revoked")) + monkeypatch.setattr( + uc, "_acting_identity", AsyncMock(return_value=("bot", {"x-api-key": "k"})) + ) + monkeypatch.setattr(sg, "_REQUIRE_LINKED_USER", False) + + principal, headers, sgp_user_id = await uc._turn_identity(_inbound()) + + assert principal == "bot" + assert headers == {"x-api-key": "k"} + assert sgp_user_id is None # <- what makes _run_turn offer a re-link + + async def test_credential_is_not_deleted_on_rejection(self, monkeypatch): + # Non-destructive on purpose: a systemic 401 costs a bot-fallback turn and a + # rate-limited nudge, then recovers by itself. Tombstoning rows would not. + from src.adapters.authentication.exceptions import AuthenticationError + + uc = SlackGatewayUseCase() + repo = MagicMock() + service = SimpleNamespace( + acting_headers=AsyncMock(return_value=self._headers()), repository=repo + ) + monkeypatch.setattr( + uc, + "_resolve_invoking_identity", + AsyncMock( + return_value=SimpleNamespace( + sgp_user_id="sgp-1", principal={}, sgp_account_id="a" + ) + ), + ) + monkeypatch.setattr(uc, "_identity_link_service", lambda: service) + self._verify(monkeypatch, raises=AuthenticationError("revoked")) + monkeypatch.setattr(uc, "_acting_identity", AsyncMock(return_value=("bot", {}))) + monkeypatch.setattr(sg, "_REQUIRE_LINKED_USER", False) + + await uc._turn_identity(_inbound()) + + repo.revoke.assert_not_called() + assert not any( + "revoke" in str(c) or "delete" in str(c) for c in repo.mock_calls + ) + + +@pytest.mark.unit +class TestUnlinkedTurnContext: + """An unlinked turn must tell the agent it isn't the user. + + Without it the turn runs on the shared bot identity and can't tell, so it reports + on ITS OWN access as though it were the asker's. Observed in production: + confidently "no Linear access, verified two ways" plus an overstated GitHub claim, + while the gateway was simultaneously DMing that person a connect link. Two + messages that contradicted each other, one of them wrong about the user. + + It also removes the reason the agent invents its own OAuth link — a URL with a + localhost redirect, which cannot work from Slack and reads like a real + instruction. + """ + + def test_unlinked_context_disclaims_personal_access(self): + text = sg._turn_content(_inbound(), "what's in my linear?", unlinked=True) + assert "NOT running as the person" in text + # Must not let the agent pass its own reach off as the user's. + assert "personal integrations" in text + assert "as if it were theirs" in text + + def test_unlinked_context_forbids_inventing_oauth_links(self): + text = sg._turn_content(_inbound(), "hi", unlinked=True) + assert "do NOT generate authorization or OAuth links" in text + # And explains why there's no need: one is already on its way. + assert "ALREADY been sent a link" in text + + def test_linked_turn_says_none_of_it(self): + text = sg._turn_content(_inbound(), "what's in my linear?", unlinked=False) + assert "NOT running as the person" not in text + assert "OAuth" not in text + + def test_channel_context_survives_either_way(self): + # The disclaimer is additive: the agent still needs the channel id to read + # thread history with its Slack tools. + for unlinked in (True, False): + text = sg._turn_content(_inbound(channel="C9"), "hi", unlinked=unlinked) + assert "channel_id=C9" in text + + def test_composes_with_the_self_posts_directive(self): + # golden-agent is both self-posting AND commonly unlinked; it needs both. + text = sg._turn_content(_inbound(), "hi", self_posts=True, unlinked=True) + assert "post_message" in text + assert "NOT running as the person" in text + + def test_user_prompt_is_still_last(self): + # The prompt must follow the context, or the agent reads the directives as + # part of the question. + text = sg._turn_content(_inbound(), "MY QUESTION", unlinked=True) + assert text.rstrip().endswith("MY QUESTION") + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestUnlinkedFlagIsWiredToIdentity: + """The flag tracks sgp_user_id — the same signal that offers the link — so the + agent is told it's unlinked exactly when the user is being asked to link, and the + two messages agree instead of contradicting each other.""" + + @staticmethod + def _acp(monkeypatch): + acp = MagicMock() + acp.agent_repository = MagicMock(get=AsyncMock(return_value=MagicMock())) + # Fails AFTER _turn_content has been built, which is all this test needs. + acp.handle_rpc_request = AsyncMock(side_effect=RuntimeError("stop here")) + monkeypatch.setattr( + "src.temporal.scheduled_agent_run_factory.build_acp_use_case_for_principal", + lambda *a, **k: acp, + ) + + @pytest.mark.parametrize( + "sgp_user_id, expect_unlinked", [("sgp-1", False), (None, True)] + ) + async def test_flag_mirrors_whether_the_turn_runs_as_a_person( + self, monkeypatch, sgp_user_id, expect_unlinked + ): + captured = {} + + def fake_turn_content(inbound, prompt, *, self_posts=False, unlinked=False): + captured["unlinked"] = unlinked + return "ctx" + + monkeypatch.setattr(sg, "_turn_content", fake_turn_content) + self._acp(monkeypatch) + + with contextlib.suppress(Exception): + await SlackGatewayUseCase()._dispatch( + sg.Target(agent_name="golden-agent", config_id=None), + _inbound(), + "hi", + {"user_id": "u"}, + {"x-api-key": "k"}, + sgp_user_id=sgp_user_id, + ) + + assert captured.get("unlinked") is expect_unlinked + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestReplayPendingTurn: + """Re-running the message that prompted a link.""" + + def _wire(self, monkeypatch): + uc = SlackGatewayUseCase() + ran = {} + + async def run_turn(inbound, *, offer_link=True): + ran["inbound"] = inbound + ran["offer_link"] = offer_link + + monkeypatch.setattr(uc, "_run_turn", run_turn) + monkeypatch.setattr(uc, "_set_status", AsyncMock()) + return uc, ran + + async def test_reconstructs_the_original_turn(self, monkeypatch): + uc, ran = self._wire(monkeypatch) + ok = await uc.replay_pending_turn( + team_id="T1", + user_id="U1", + pending_turn={ + "text": "what's in my linear?", + "channel": "C9", + "thread_ts": "1700.1", + }, + ) + assert ok + inbound = ran["inbound"] + assert (inbound.team_id, inbound.user) == ("T1", "U1") + assert (inbound.channel, inbound.thread_ts) == ("C9", "1700.1") + assert inbound.text == "what's in my linear?" + + async def test_selector_is_rederived_like_normalize(self, monkeypatch): + # A selector-driven turn must resolve to the same target it would have; the + # nonce stores only the text, so the selector is re-derived the same way. + uc, ran = self._wire(monkeypatch) + await uc.replay_pending_turn( + team_id="T1", + user_id="U1", + pending_turn={"text": "some-config summarise this", "channel": "C1"}, + ) + assert ran["inbound"].selector == "some-config" + + async def test_never_offers_another_link(self, monkeypatch): + # The loop guard. This turn exists BECAUSE they just linked; offering again + # would mint a nonce, DM a link, and invite the same loop on the next click. + uc, ran = self._wire(monkeypatch) + await uc.replay_pending_turn( + team_id="T1", user_id="U1", pending_turn={"text": "hi", "channel": "C1"} + ) + assert ran["offer_link"] is False + + async def test_sets_the_thinking_indicator(self, monkeypatch): + # The answer lands minutes after the user clicked a web page, so without this + # a reply appears out of nowhere. + uc, _ran = self._wire(monkeypatch) + await uc.replay_pending_turn( + team_id="T1", user_id="U1", pending_turn={"text": "hi", "channel": "C1"} + ) + uc._set_status.assert_awaited_once() + + @pytest.mark.parametrize( + "pending", + [ + None, + {}, + {"text": "", "channel": "C1"}, + {"text": "hi", "channel": ""}, + {"text": " ", "channel": "C1"}, + ], + ) + async def test_incomplete_pending_turn_does_nothing(self, monkeypatch, pending): + uc, ran = self._wire(monkeypatch) + assert ( + await uc.replay_pending_turn( + team_id="T1", user_id="U1", pending_turn=pending + ) + is False + ) + assert ran == {} + + async def test_missing_identity_does_nothing(self, monkeypatch): + uc, ran = self._wire(monkeypatch) + assert ( + await uc.replay_pending_turn( + team_id="", user_id="U1", pending_turn={"text": "hi", "channel": "C1"} + ) + is False + ) + assert ran == {} + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestOfferLinkSuppression: + async def test_run_turn_skips_the_offer_when_told_to(self, monkeypatch): + uc = SlackGatewayUseCase() + offered = [] + monkeypatch.setattr( + uc, "_offer_link", AsyncMock(side_effect=lambda i: offered.append(i)) + ) + # Unlinked: normally this is exactly when an offer fires. + monkeypatch.setattr( + uc, + "_turn_identity", + AsyncMock(return_value=("bot", {"x-api-key": "k"}, None)), + ) + monkeypatch.setattr( + uc, "_resolve_target", AsyncMock(side_effect=RuntimeError("stop")) + ) + monkeypatch.setattr(uc, "_deliver", AsyncMock()) + + await uc._run_turn(_inbound(), offer_link=False) + assert offered == [] + + await uc._run_turn(_inbound(), offer_link=True) + assert len(offered) == 1