fix(craft): record receipts for ALWAYS-policy writes

The ALWAYS branch forwards without session resolution, so an admin-relaxed
write executed with no receipt at all. A write now resolves its session
best-effort and records, reads keep the cheap path, and an unattributable
write goes unrecorded rather than blocked.

Author: nmgarza5

backend/onyx/sandbox_proxy/addons/gate.py

@@ -26,6 +26,7 @@
 from onyx.configs.constants import NotificationType
 from onyx.db.engine.sql_engine import get_session_with_tenant
 from onyx.db.enums import (
+    ActionEffect,
     ApprovalDecidedVia,
     ApprovalDecision,
     EndpointPolicy,
@@ -520,6 +521,39 @@ async def _dispatch_approved_request(
             )
             await self._record_receipts(flow, ctx, matched_actions, approval_id)
 
+    async def _record_always_write_receipts(
+        self,
+        flow: http.HTTPFlow,
+        sandbox: ResolvedSandbox,
+        matched_actions: AllMatchedActions,
+    ) -> None:
+        """An admin-set ALWAYS policy must not exempt a write from the
+        activity record. Best-effort: the ALWAYS path deliberately skips
+        session resolution, so a write resolves it here and an unattributable
+        one goes unrecorded rather than blocked."""
+        if not any(
+            action.effect is ActionEffect.WRITE for action in matched_actions.actions
+        ):
+            return
+        try:
+            session_id = self._resolve_gated_session(flow, sandbox)
+        except Exception:
+            logger.exception(
+                "receipt_session_lookup_error tenant=%s sandbox=%s host=%s",
+                sandbox.tenant_id,
+                sandbox_log_label(sandbox),
+                flow.request.host,
+            )
+            return
+        if session_id is None:
+            return
+        await self._record_receipts(
+            flow,
+            sandbox.with_session(session_id),
+            matched_actions,
+            approval_id=None,
+        )
+
     async def _record_receipts(
         self,
         flow: http.HTTPFlow,
@@ -766,6 +800,7 @@ async def _resolve_and_match(
                     ),
                     credential_outcome_label(injection),
                 )
+                await self._record_always_write_receipts(flow, sandbox, matched_actions)
             return None
 
         # ASK: resolve the originating session before prompting. An

backend/tests/unit/sandbox_proxy/test_gate_receipts.py

@@ -195,3 +195,74 @@ def test_hooks_ignore_unrecorded_flows(
     asyncio.run(addon.error(tflow.tflow()))
 
     assert calls == []
+
+
+def test_always_policy_writes_still_record(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    addon = _addon()
+    recorded: list[Any] = []
+
+    async def _fake_record(*args: Any, **kwargs: Any) -> None:
+        recorded.append((args, kwargs))
+
+    session_id = uuid4()
+    monkeypatch.setattr(addon, "_record_receipts", _fake_record)
+    monkeypatch.setattr(
+        addon, "_resolve_gated_session", lambda _flow, _sandbox: session_id
+    )
+    flow = tflow.tflow()
+    sandbox = _ctx().without_session()
+
+    asyncio.run(addon._record_always_write_receipts(flow, sandbox, _MATCHED))
+
+    assert len(recorded) == 1
+    ctx = recorded[0][0][1]
+    assert ctx.session_id == session_id
+
+
+def test_always_policy_reads_skip_session_resolution(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    addon = _addon()
+
+    def _must_not_resolve(_flow: Any, _sandbox: Any) -> None:
+        raise AssertionError("reads must not pay session resolution")
+
+    monkeypatch.setattr(addon, "_resolve_gated_session", _must_not_resolve)
+    read_only = AllMatchedActions(
+        actions=(
+            MatchedAction(
+                action_type="slack.messages.read",
+                display_name="Read messages",
+                description="Read messages.",
+                policy=EndpointPolicy.ALWAYS,
+                effect=ActionEffect.READ,
+            ),
+        ),
+        target=GatedTarget(kind=GatedAppKind.EXTERNAL_APP, id=1, app_name="Slack"),
+    )
+
+    asyncio.run(
+        addon._record_always_write_receipts(
+            tflow.tflow(), _ctx().without_session(), read_only
+        )
+    )
+
+
+def test_always_policy_unattributable_write_goes_unrecorded(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    addon = _addon()
+
+    async def _must_not_record(*_args: Any, **_kwargs: Any) -> None:
+        raise AssertionError("unattributable writes record nothing")
+
+    monkeypatch.setattr(addon, "_record_receipts", _must_not_record)
+    monkeypatch.setattr(addon, "_resolve_gated_session", lambda _flow, _sandbox: None)
+
+    asyncio.run(
+        addon._record_always_write_receipts(
+            tflow.tflow(), _ctx().without_session(), _MATCHED
+        )
+    )