From 151b4a64d2b10b16d81da2e824a20b4fb78372a0 Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Wed, 19 Aug 2026 12:28:57 +0300 Subject: [PATCH 1/3] =?UTF-8?q?feat(workspace):=20addressed=20asks=20?= =?UTF-8?q?=E2=80=94=20the=20column,=20its=20validation,=20and=20the=20thr?= =?UTF-8?q?ee=20renderings=20(ent#364)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The OSS half of ent#364. An agent raises a question, an approval request or an update addressed to a specific workspace user; it renders in three places and is cleared everywhere when answered. **One row, not three.** An ask IS an `operator_queue` row with an addressee — `operator_queue.addressed_to_email`, nullable, NULL meaning "the operator answers" so every existing row keeps its meaning. Both migration tracks (invariant #3): `migrations.py::operator_queue_addressed_to` and Alembic `0039_operator_queue_addressed_to`. No new table, because ent#164 builds a hard human-approval gate on these same rows and two approval systems is the failure mode. **Why the addressee is a column and not a key in `context`.** It decides who may answer an ask and whose sidebar it appears in — an authorization decision — while `context` is free-form agent-authored JSON whose hygiene clamp bounds size and type only. Put it there and an agent picks its own audience. So it is validated at that same #1632 ingestion boundary (`_validated_addressee`) against `agent_on_roster(agent, email, include_owned=False)`: an agent may address only someone already on its roster. `include_owned=False` is the rule, not a default — flipping it on would hand a client agents nobody shared with them. Fails closed throughout: a non-string, an address-shaped-nothing, an over-length value, an off-roster address, or a roster read that raises all resolve to NULL — an operator ask. Malformed values are refused BEFORE the lookup, so garbage cannot amplify into a DB read per item. A `context.addressed_to_email` is never honoured; pinned by a test, since that is precisely the hole the column exists to close. **Three renderings, one list.** `PortalAsks.vue` is used by the agent page and inline above the chat composer, and the sidebar shows `store.askCount` — all from a single `fetchAsks()`. That is what makes "answering anywhere clears it everywhere" true with no sync step: three bespoke queries is how it stops being true. Answering removes the row from local state rather than patching it, so a stale "pending" copy can never offer to answer the same ask twice. An expired ask RENDERS as expired and is excluded from the count: #1142 deletes terminal rows, and an ask that simply vanishes reads as "answered" to the person who did not answer it — but it is not something to nag about either. **Poll, not a socket.** The Workspace has no WebSocket (`operator_queue_new` goes out on the platform `/ws`, which a portal client is not on) and already polls elsewhere; 20s, visibility-aware, torn down with the view. An ask is a human-latency decision, and new transport for one badge is not worth it. The gated read/answer surface is the enterprise `workspace_asks` module; this half is the edition-agnostic primitive plus the UI, and the UI self-hides when the endpoint 404/403s so an OSS build renders nothing and logs nothing. 16 OSS tests (validation, fail-closed paths, no-context-honouring, column round-trip) + 9 frontend (one-list contract, answering clears every surface, expired-renders-uncounted, silent-on-unentitled, refusal surfaced). Full frontend suite 937 passed; `check_alembic_heads` reports 1 head. Related to ent#364 --- docs/memory/architecture.md | 3 + src/backend/db/migrations.py | 24 +++ src/backend/db/operator_queue.py | 7 + src/backend/db/schema.py | 1 + src/backend/db/tables.py | 4 + .../0039_operator_queue_addressed_to.py | 33 ++++ .../services/operator_queue_service.py | 54 ++++- .../src/components/portal/PortalAgentPage.vue | 7 + .../src/components/portal/PortalAsks.vue | 121 ++++++++++++ .../components/portal/PortalConversation.vue | 13 ++ .../src/components/portal/PortalSidebar.vue | 19 +- src/frontend/src/stores/clientPortal.js | 62 ++++++ src/frontend/src/views/Portal.vue | 32 ++- src/frontend/tests/unit/workspaceAsks.spec.js | 184 ++++++++++++++++++ tests/unit/test_ent364_addressed_asks.py | 144 ++++++++++++++ 15 files changed, 704 insertions(+), 4 deletions(-) create mode 100644 src/backend/migrations/versions/0039_operator_queue_addressed_to.py create mode 100644 src/frontend/src/components/portal/PortalAsks.vue create mode 100644 src/frontend/tests/unit/workspaceAsks.spec.js create mode 100644 tests/unit/test_ent364_addressed_asks.py diff --git a/docs/memory/architecture.md b/docs/memory/architecture.md index 3bf51f31ae..ec49eaa642 100644 --- a/docs/memory/architecture.md +++ b/docs/memory/architecture.md @@ -1235,6 +1235,8 @@ Bulk ops scope writes to the caller's accessible agents (tri-state: admin = no f WebSocket events: `operator_queue_new`, `operator_queue_responded`, `operator_queue_acknowledged`, `operator_queue_cleared` (one per bulk op, #1017; `notifications_cleared` for the notifications variant). Backed by the 5s Operator Queue Sync background service. +**Addressed asks (ent#364).** `operator_queue.addressed_to_email` names the HUMAN an item is for; `agent_name` only ever said which agent. NULL — every pre-ent#364 row — means "the operator answers", so the column is additive by construction. It is a **column and not a key in `context`** because it is an authorization decision (who may answer, whose Workspace sidebar it appears in) and `context` is agent-authored JSON whose clamp bounds size and type only. Validated at the one agent-authored boundary (`operator_queue_service._validated_addressee`, inside the #1632 clamp) against `client_portal.service.agent_on_roster(agent, email, include_owned=False)` — an agent may address only someone already on its roster — and **fails closed**: malformed, off-roster, or an unreadable roster all drop to NULL. The read/answer surface is the entitled `workspace_asks` module (`/api/enterprise/client-portal/asks`), which re-checks the roster at read time so a revoked share stops showing an already-raised ask, projects an explicit client-facing shape (never `context`), and answers through the OSS respond path so write-back, audit fields and the WS broadcast are unforked. `responded_by_id` stays NULL for a client answerer — there is no `users` row, and inventing one would be a lie in the audit trail. + **Ingestion caps (#1632).** There is **no HTTP create** — every operator-queue item is created through `db.create_operator_queue_item`, and the only untrusted producer is the sync service reading the agent-authored `~/.trinity/operator-queue.json` (`operator_queue_service._sync_agent`). #1402 makes this queue the approval channel for irreversible actions, so a compromised / prompt-injected agent that floods plausible "approve this" items causes operator fatigue → reflexive approval (XSS is already handled by DOMPurify; the exposure is volume + social engineering). The one agent-authored seam is bounded by two independent limits plus field hygiene, all env-tunable (see requirements §26.7): (1) a per-agent **pending-DEPTH cap** (`db.count_operator_queue_pending_for_agent`, default 25) — the **primary** bound, DB-measured ⇒ Redis-independent; at the cap the sync **stops** ingesting (never drips a growing file) and holds the surplus behind one aggregated alert; (2) a per-agent + fleet **create RATE limit** (`rate_limiter.check`, 60/60s + 300/60s, fail-open); (3) a total field-hygiene clamp (`_clamp_ingested_item`, inside the #1525 create try/except) — truncate-with-marker title/question, context/options over cap → marker (validated `execution_id` only), non-dict context → `{}`, `created_at` normalized to ingest time, priority validate-only; (4) a **reserved-id guard** rejecting agent ids that start with a platform-reserved prefix (`queue-flood-`/`poison-`/`cb-dormant-`/`sync-failing-`/`git-bloat-`/`skill-not-found-`/`val_`/`system-seed-`/`base-image-stale-`/`alert-budget-`) so an agent can't pre-create — and via `on_conflict_do_nothing` silence — its own flood alarm or the #1402 poison alert, plus a malformed/oversize-id reject; (5) a per-cycle scan bound (skip an oversized file wholesale; cap at 500 requests/cycle). The 5s sync loop is **leader-locked** (`opqueue:leader`, mirror monitoring #1464) so `--workers 2` doesn't double-charge / double-broadcast / double-scan. Platform creates bypass `_sync_agent`, and the exemption is **scoped by influence, not by caller location (#1677)**: a *platform-only* emitter (volume bound by platform cadence — edge-triggered, idempotent/bucketed id, leader-locked, or operator-driven: lease-reaper poison-park #1402, `validation_service._notify_operator_on_failure` — converted to a **direct DB create** in #1632 — and the internal breaker/skill/git alerts) stays direct and unthrottled, while an *agent-influenceable* one — `task_execution_service._alert_skill_not_found` (#1410), whose per-command dedup distinct unknown slash-commands defeat at $0/turn — routes through `operator_queue_service.create_bounded_alert`: a per-(agent, registered-type) pending-DEPTH budget (`OPERATOR_ALERT_MAX_PENDING_PER_TYPE`, default 5; type derived from `item["type"]` against the `_BUDGETED_ALERT_TYPES` frozenset; **fail-closed on every arm** — an unreadable count / unregistered type / failed create suppresses the ALERT only, the FAILED execution rows stay the primary surface, and the paired notification is gated on the same bool), with ONE cooldown-gated `alert-budget-{agent}-{type}-b{bucket}` episode alert per window (deterministic bucket ⇒ the `(agent_name, request_id)` on-conflict target dedups cross-worker; no `held` count, no agent-controlled text). The classification is CI-forced by the AST caller-parity guard `tests/unit/test_1677_operator_alert_emitters.py` (every `create_operator_queue_item` call site must be allowlisted platform-only or routed; OSS tree only — the private submodule owns its twin); a sink-level default bound was rejected because it fails QUIETLY at the load-bearing poison-park create where the parity test fails LOUDLY at CI (#1890's lesson deliberately inverted). A generous DB-sink belt in `create_item` (reject title>4 KiB / question>16 KiB / context>64 KiB / id>512; the derived `execution_id` COLUMN belted to None on non-str/>512, #1677) is a second layer so the exemption is never solely load-bearing (#1525 validate-at-boundary-AND-at-sink). A depth-held / rate-skipped / oversize-file episode emits **one** `queue-flood-{agent}-{utc_now_iso()}` alert (un-guessable id, priority `high`, in-memory cooldown, emit-failure-safe). The platform-emitter residual named on the pull-mode default-ON gate list (#1081 / `TARGET_ARCHITECTURE.md`) is closed by #1677; the remaining gate items are unchanged. ### Platform Audit Log (SEC-001) @@ -2194,6 +2196,7 @@ CREATE TABLE operator_queue ( responded_at TEXT, acknowledged_at TEXT, cleared_at TEXT, -- #1017: NULL = visible; set = hidden by Clear All (rows deleted by the #1142 retention sweep past operator_queue_retention_days) + addressed_to_email TEXT, -- ent#364: the human this ask is for; NULL = operator ask. Validated at ingestion against the agent's roster, never trusted from the payload FOREIGN KEY (responded_by_id) REFERENCES users(id) ); CREATE INDEX idx_operator_queue_agent ON operator_queue(agent_name); diff --git a/src/backend/db/migrations.py b/src/backend/db/migrations.py index f625a2446c..cb428102f5 100644 --- a/src/backend/db/migrations.py +++ b/src/backend/db/migrations.py @@ -3419,6 +3419,29 @@ def _migrate_portal_session_resume(cursor, conn): conn.commit() +def _migrate_operator_queue_addressed_to(cursor, conn): + """Address an operator-queue item to a specific human (ent#364). + + `agent_name` says WHICH AGENT an item belongs to; nothing said which person + should answer it. Workspace asks need that, and it has to be a column rather + than a key inside `context`: `context` is agent-authored free-form JSON, so + burying the addressee there would let an agent decide who may answer and whose + sidebar an ask appears in. The value is validated at the ingestion boundary + (`operator_queue_service`) against the agent's roster. + + Nullable with no default, so every existing row keeps meaning exactly what it + meant: an ask for the operator. + """ + _safe_add_column( + cursor, + "operator_queue", + "addressed_to_email", + "ALTER TABLE operator_queue ADD COLUMN addressed_to_email TEXT", + log_msg="Adding addressed_to_email to operator_queue for workspace asks (ent#364)", + ) + conn.commit() + + def _migrate_portal_chat_state(cursor, conn): """Per-user star + read cursor for Workspace chats (ent#359). @@ -3556,4 +3579,5 @@ def _migrate_portal_chat_state(cursor, conn): ("client_portal_tables_to_oss", _migrate_client_portal_tables_to_oss), ("portal_session_resume", _migrate_portal_session_resume), ("portal_chat_state", _migrate_portal_chat_state), + ("operator_queue_addressed_to", _migrate_operator_queue_addressed_to), ] diff --git a/src/backend/db/operator_queue.py b/src/backend/db/operator_queue.py index d9c0926bbb..86a6aff9ad 100644 --- a/src/backend/db/operator_queue.py +++ b/src/backend/db/operator_queue.py @@ -94,6 +94,7 @@ def _row_to_item(row) -> Dict: "responded_at": row["responded_at"], "acknowledged_at": row["acknowledged_at"], "cleared_at": row["cleared_at"], # #1017 + "addressed_to_email": row["addressed_to_email"], # ent#364 } # Columns selected for a full queue-item record, in the canonical order. @@ -118,6 +119,7 @@ def _row_to_item(row) -> Dict: operator_queue.c.responded_at, operator_queue.c.acknowledged_at, operator_queue.c.cleared_at, # #1017 — Clear All hide flag + operator_queue.c.addressed_to_email, # ent#364 — the human it is for ) def create_item(self, agent_name: str, item: Dict) -> str: @@ -206,6 +208,11 @@ def create_item(self, agent_name: str, item: Dict) -> str: execution_id=context_execution_id, created_at=item.get("created_at") or utc_now_iso(), expires_at=item.get("expires_at"), + # ent#364: already validated against the agent's roster by + # `operator_queue_service._validated_addressee`. This layer stores it; + # it does not decide it, and it must never derive it from `context` + # (which is agent-authored). + addressed_to_email=item.get("addressed_to_email"), ).on_conflict_do_nothing(index_elements=["agent_name", "request_id"]) # Insert + re-read in one transaction: on conflict the insert is a no-op diff --git a/src/backend/db/schema.py b/src/backend/db/schema.py index c35bd57fdc..43ea0ee088 100644 --- a/src/backend/db/schema.py +++ b/src/backend/db/schema.py @@ -1261,6 +1261,7 @@ responded_at TEXT, acknowledged_at TEXT, cleared_at TEXT, + addressed_to_email TEXT, FOREIGN KEY (responded_by_id) REFERENCES users(id) ) """, diff --git a/src/backend/db/tables.py b/src/backend/db/tables.py index d6b883c1d3..3a15201454 100644 --- a/src/backend/db/tables.py +++ b/src/backend/db/tables.py @@ -1094,6 +1094,10 @@ def process_bind_param(self, value, dialect): Column("responded_at", Text), Column("acknowledged_at", Text), Column("cleared_at", Text), # #1017 — Clear All hide flag + # ent#364: the human this item is addressed to. NULL = an operator ask + # (every pre-ent#364 row). Validated at ingestion against the agent's + # roster — never trusted from the agent-authored payload. + Column("addressed_to_email", Text), ) nevermined_agent_config = Table( diff --git a/src/backend/migrations/versions/0039_operator_queue_addressed_to.py b/src/backend/migrations/versions/0039_operator_queue_addressed_to.py new file mode 100644 index 0000000000..fd90284eb7 --- /dev/null +++ b/src/backend/migrations/versions/0039_operator_queue_addressed_to.py @@ -0,0 +1,33 @@ +"""operator_queue.addressed_to_email — address an ask to a human (ent#364) + +`agent_name` says which AGENT an item belongs to; nothing said which PERSON should +answer it. Workspace asks need that. + +It is a column and not a key inside `context` on purpose: `context` is +agent-authored free-form JSON whose hygiene clamp bounds size and type only, so +putting the addressee there would let an agent decide who may answer an ask and +whose sidebar it appears in. The value is validated at the ingestion boundary +(`services/operator_queue_service.py`) against the agent's roster. + +Nullable, no default: every existing row keeps meaning what it meant — an ask for +the operator. + +Revision ID: 0039_operator_queue_addressed_to +Revises: 0038_portal_chat_state +Create Date: 2026-08-19 +""" +import sqlalchemy as sa +from alembic import op + +revision = "0039_operator_queue_addressed_to" +down_revision = "0038_portal_chat_state" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("operator_queue", sa.Column("addressed_to_email", sa.Text(), nullable=True)) + + +def downgrade() -> None: + op.drop_column("operator_queue", "addressed_to_email") diff --git a/src/backend/services/operator_queue_service.py b/src/backend/services/operator_queue_service.py index 09ebe50158..28b9995855 100644 --- a/src/backend/services/operator_queue_service.py +++ b/src/backend/services/operator_queue_service.py @@ -83,6 +83,9 @@ OPERATOR_QUEUE_OPTIONS_MAX_BYTES = int(os.getenv("OPERATOR_QUEUE_OPTIONS_MAX_BYTES", "4096")) OPERATOR_QUEUE_ID_MAX = int(os.getenv("OPERATOR_QUEUE_ID_MAX", "256")) OPERATOR_QUEUE_EXECUTION_ID_MAX = int(os.getenv("OPERATOR_QUEUE_EXECUTION_ID_MAX", "128")) +# ent#364: RFC 5321 caps an address at 320 chars; anything longer is not an +# email and never matches a roster row, so it is refused before the DB read. +OPERATOR_QUEUE_EMAIL_MAX = int(os.getenv("OPERATOR_QUEUE_EMAIL_MAX", "320")) # Flood alert: one per episode, un-guessable id, in-memory cooldown. OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS = int( os.getenv("OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS", "300") @@ -189,7 +192,50 @@ def _truncate_with_marker(text: str, max_len: int) -> str: return text[:keep] + _TRUNC_MARKER -def _clamp_ingested_item(req: dict) -> dict: +def _validated_addressee(agent_name: str, raw) -> Optional[str]: + """The email an ask is addressed to, or None (ent#364). + + This is the one field on an agent-authored item that is an AUTHORIZATION + decision: it determines who may answer the ask and whose Workspace sidebar it + appears in. So it is validated here, at the same boundary that clamps every + other agent-supplied field, against the agent's own roster — an agent may + address only someone it was already shared with. + + `include_owned=False` is the rule, not a default (see `agent_on_roster`'s + docstring): an external client's scope is exactly what was shared with them, + and the owned-agents branch would hand a client agents nobody gave them. + + Fails CLOSED: a malformed value, an off-roster address, or a roster lookup that + raises all yield None — an operator ask — because addressing an ask we could + not validate is worse than not addressing it. Never raises, per this module's + #1632 contract. + """ + if not isinstance(raw, str): + return None + email = raw.strip().lower() + if not email or len(email) > OPERATOR_QUEUE_EMAIL_MAX or "@" not in email: + return None + try: + from client_portal.service import agent_on_roster + + if agent_on_roster(agent_name, email, include_owned=False): + return email + logger.info( + "[OperatorQueue] %s addressed an item to an off-roster email; " + "treating it as an operator ask (ent#364)", + agent_name, + ) + except Exception: # noqa: BLE001 — a roster read must never break ingestion + logger.warning( + "[OperatorQueue] could not validate the addressee for %s; " + "treating it as an operator ask (ent#364)", + agent_name, + exc_info=True, + ) + return None + + +def _clamp_ingested_item(req: dict, agent_name: str = "") -> dict: """#1632: total field-hygiene clamp for an agent-authored queue item. Called INSIDE the create try/except so any failure is quarantined by #1525 @@ -246,6 +292,10 @@ def _clamp_ingested_item(req: dict) -> dict: if out.get("priority") not in _VALID_PRIORITIES: out["priority"] = "medium" + # ent#364: the addressee is an authorization decision, so it is resolved here + # rather than trusted. Absent/invalid/off-roster → None, i.e. an operator ask. + out["addressed_to_email"] = _validated_addressee(agent_name, out.get("addressed_to_email")) + # Ignore the agent-supplied created_at (a future date pins the item atop the # `created_at DESC` sort, C4). expires_at is honored as authored. out["created_at"] = utc_now_iso() @@ -767,7 +817,7 @@ async def _sync_agent(self, agent_name: str): # New item — clamp then create. The clamp runs INSIDE the try so any # clamp/create failure is quarantined by #1525 rather than hot-looping. try: - clamped = _clamp_ingested_item(req) + clamped = _clamp_ingested_item(req, agent_name) db.create_operator_queue_item(agent_name, clamped) admitted += 1 new_items.append(clamped) diff --git a/src/frontend/src/components/portal/PortalAgentPage.vue b/src/frontend/src/components/portal/PortalAgentPage.vue index adf423f8fa..308cce5546 100644 --- a/src/frontend/src/components/portal/PortalAgentPage.vue +++ b/src/frontend/src/components/portal/PortalAgentPage.vue @@ -37,6 +37,12 @@ >Start a chat + +
+ +
+
@@ -375,6 +381,7 @@ import InlineError from '@/components/InlineError.vue' import LoadFailed from '@/components/LoadFailed.vue' import ReportRenderer from '@/components/reports/ReportRenderer.vue' import ReportSummary from '@/components/reports/ReportSummary.vue' +import PortalAsks from '@/components/portal/PortalAsks.vue' import PortalAvatar from './PortalAvatar.vue' import { PORTAL_BUCKET_LABELS, availabilityChip } from './portalUtils' diff --git a/src/frontend/src/components/portal/PortalAsks.vue b/src/frontend/src/components/portal/PortalAsks.vue new file mode 100644 index 0000000000..da50a22723 --- /dev/null +++ b/src/frontend/src/components/portal/PortalAsks.vue @@ -0,0 +1,121 @@ + + + + diff --git a/src/frontend/src/components/portal/PortalConversation.vue b/src/frontend/src/components/portal/PortalConversation.vue index bc300c711c..94fb462974 100644 --- a/src/frontend/src/components/portal/PortalConversation.vue +++ b/src/frontend/src/components/portal/PortalConversation.vue @@ -128,6 +128,16 @@
+ +
+
+ +
+
+
@@ -257,6 +267,7 @@ import { agentDisplayName } from '@/utils/agentName' import PortalAvatar from './PortalAvatar.vue' import PortalStarButton from './PortalStarButton.vue' import PortalTypeahead from './PortalTypeahead.vue' +import PortalAsks from './PortalAsks.vue' import { deliveryFailureReason, mentionedAgents, @@ -306,6 +317,8 @@ const props = defineProps({ const emit = defineEmits(['switch-agent', 'session-adopted', 'sessions-changed', 'open-files', 'open-menu', 'toggle-star', 'escalate-to-room']) const store = useClientPortalStore() +// ent#364: one list, filtered — never a second fetch for this surface. +const agentAsks = computed(() => store.asksForAgent(props.agent)) const messages = ref([]) const currentSessionId = ref(props.sessionId) const loadingHistory = ref(false) diff --git a/src/frontend/src/components/portal/PortalSidebar.vue b/src/frontend/src/components/portal/PortalSidebar.vue index 46a5a3ac9a..2207e9138b 100644 --- a/src/frontend/src/components/portal/PortalSidebar.vue +++ b/src/frontend/src/components/portal/PortalSidebar.vue @@ -6,9 +6,20 @@
Workspace + + {{ askCount > 99 ? '99+' : askCount }} {{ totalWaiting > 99 ? '99+' : totalWaiting }}
@@ -216,6 +227,7 @@ import { computed, ref } from 'vue' import PortalAvatar from './PortalAvatar.vue' import ChatRow from './PortalChatRow.vue' import BaseBadge from '@/components/base/BaseBadge.vue' +import { useClientPortalStore } from '@/stores/clientPortal' import { groupThreadsByDate, partitionStarred, unreadByAgent, totalUnread, availabilityChip, signOutLabelFor, @@ -251,6 +263,11 @@ const grouped = computed(() => groupThreadsByDate(split.value.rest)) const waiting = computed(() => unreadByAgent(props.threads)) const waitingFor = (name) => waiting.value[name] || 0 const totalWaiting = computed(() => totalUnread(props.threads)) +// ent#364: read from the store rather than taken as a prop, because the count and +// the two ask renderings must come from ONE list — a prop threaded from the view +// would be a second path to the same fact, free to disagree with it. +const asksStore = useClientPortalStore() +const askCount = computed(() => asksStore.askCount) // A row key has to include the kind: thread ids and room ids are independent // spaces, so two chats of different kinds could collide on a bare id. diff --git a/src/frontend/src/stores/clientPortal.js b/src/frontend/src/stores/clientPortal.js index cc635b6c7e..6b42835863 100644 --- a/src/frontend/src/stores/clientPortal.js +++ b/src/frontend/src/stores/clientPortal.js @@ -99,6 +99,12 @@ export const useClientPortalStore = defineStore('clientPortal', { // re-appearing, which is indistinguishable from "you were never signed in" // and reads as the app having lost the thread. sessionExpired: false, + // ent#364 — agent-initiated asks addressed to this user. ONE list, read by all + // three surfaces (sidebar count, agent page, inline in chat), because the + // underlying row is one row: answering anywhere clears it everywhere with no + // sync step, and three separate queries is how that stops being true. + asks: [], + asksAvailable: false, // false when the module is absent/unentitled (404/403) // Where the user was when it expired, so re-authenticating returns them // there instead of the roster root. resumePath: null, @@ -187,6 +193,16 @@ export const useClientPortalStore = defineStore('clientPortal', { isClientSignedIn() { return !!this.portalToken || this.isPlatformSession }, + // ent#364: only PENDING asks are actionable. An expired one still renders (see + // `asks`), but it is not something the badge should nag about — the person did + // not fail to answer something they can still answer. + openAsks: (state) => state.asks.filter((a) => a.status === 'pending'), + askCount() { + return this.openAsks.length + }, + asksForAgent() { + return (agentName) => this.asks.filter((a) => a.agent_name === agentName) + }, }, actions: { @@ -952,6 +968,52 @@ export const useClientPortalStore = defineStore('clientPortal', { return kept }, + // ent#364 — the ONE read all three surfaces use. + // + // Uses the bare `axios` export like every other call in this store. #2261 is + // moving this whole file onto a dedicated instance (`portalHttp`) so a + // workspace request can never inherit the platform JWT from + // `axios.defaults`; when that lands, these two calls move with the other 29. + // Deliberately NOT pre-empting it here: half the file on one transport and + // half on another is worse than either, and that PR owns the migration. + // + // Degrades to silence: the module is enterprise-gated, so an OSS or unentitled + // build answers 404/403 and there is nothing to render. That is not an error + // worth showing a client — `asksAvailable` stays false and every surface + // renders nothing, which is how this ships in the OSS bundle at all. + async fetchAsks(agentName = null) { + if (!this.isClientSignedIn) return [] + try { + const { data } = await axios.get('/api/enterprise/client-portal/asks', { + headers: this.authHeader, + params: agentName ? { agent_name: agentName } : {}, + }) + this.asks = Array.isArray(data) ? data : [] + this.asksAvailable = true + return this.asks + } catch (err) { + this.asksAvailable = false + if (![403, 404].includes(err.response?.status)) { + console.warn('[workspace] asks unavailable:', err?.message || err) + } + this.asks = [] + return [] + } + }, + + // Answer one ask. The row is removed from local state on success rather than + // patched: the server's answer is authoritative, and a client that keeps a + // stale "pending" copy would offer to answer it twice. + async answerAsk(askId, { response = null, responseText = null } = {}) { + const { data } = await axios.post( + `/api/enterprise/client-portal/asks/${askId}/answer`, + { response, response_text: responseText }, + { headers: this.authHeader }, + ) + this.asks = this.asks.filter((a) => a.id !== askId) + return data + }, + async fetchRoster() { this.loading = true this.error = null diff --git a/src/frontend/src/views/Portal.vue b/src/frontend/src/views/Portal.vue index 2cbd96070a..908fbb8110 100644 --- a/src/frontend/src/views/Portal.vue +++ b/src/frontend/src/views/Portal.vue @@ -737,9 +737,35 @@ function resolveAgentQuery() { return true } +// ent#364 — one poll feeds all three ask renderings. +// +// The Workspace has no WebSocket (`operator_queue_new` is broadcast on the platform +// `/ws`, which a portal client is not on), and it already polls elsewhere, so a +// second live channel for this would be new transport for one badge. 20s: an ask is +// a human-latency decision, not a stream. +const ASKS_POLL_MS = 20000 +let asksTimer = null + +function startAsksPoll() { + stopAsksPoll() + if (!store.isClientSignedIn) return + store.fetchAsks() + asksTimer = setInterval(() => { + // Visibility-aware: a backgrounded tab polls nothing. The next foreground + // tick catches up, and an ask that arrived meanwhile is not lost — it is a + // row, not an event. + if (document.visibilityState === 'visible') store.fetchAsks() + }, ASKS_POLL_MS) +} + +function stopAsksPoll() { + if (asksTimer) { clearInterval(asksTimer); asksTimer = null } +} + async function bootstrap() { await store.fetchRoster() await refreshThreads() + startAsksPoll() const sid = route.params.sessionId if (sid) { const known = threads.value.find((t) => (t.id || t.session_id) === sid) @@ -753,7 +779,11 @@ async function bootstrap() { } onMounted(async () => { if (store.isClientSignedIn) await bootstrap() }) -onBeforeUnmount(() => { clearInterval(resendTimer); clearTimeout(searchTimer) }) +onBeforeUnmount(() => { + stopAsksPoll() // ent#364 — the poll must not outlive the view + clearInterval(resendTimer) + clearTimeout(searchTimer) +}) // #2258: true from the click until the credential is gone and the route has // moved. Gates the template's first branch so neither principal sees a state diff --git a/src/frontend/tests/unit/workspaceAsks.spec.js b/src/frontend/tests/unit/workspaceAsks.spec.js new file mode 100644 index 0000000000..63698bbcd6 --- /dev/null +++ b/src/frontend/tests/unit/workspaceAsks.spec.js @@ -0,0 +1,184 @@ +/** + * The Workspace asks store contract (ent#364). + * + * The feature's central claim is "one row, three renderings, cleared everywhere". + * In the client that reduces to: all three surfaces read ONE list, and answering + * removes the row from it. So these tests are about the list — the three components + * are template work over what is pinned here. + * + * Also pinned: the module is enterprise-gated, so an OSS or unentitled build gets + * 404/403 on every poll. That has to be silent, or every OSS install logs a warning + * every 20 seconds for a feature it does not have. + */ +import { describe, it, expect, beforeEach, vi } from 'vitest' +import { setActivePinia, createPinia } from 'pinia' + +vi.hoisted(() => { + const mem = new Map() + globalThis.localStorage = { + getItem: (k) => (mem.has(k) ? mem.get(k) : null), + setItem: (k, v) => mem.set(k, String(v)), + removeItem: (k) => mem.delete(k), + clear: () => mem.clear(), + } + const session = new Map() + globalThis.sessionStorage = { + getItem: (k) => (session.has(k) ? session.get(k) : null), + setItem: (k, v) => session.set(k, String(v)), + removeItem: (k) => session.delete(k), + clear: () => session.clear(), + } + globalThis.window = globalThis.window || { location: { pathname: '/workspace' } } +}) + +vi.mock('@/stores/auth', () => ({ + useAuthStore: () => ({ isAuthenticated: false, authHeader: {}, logout: vi.fn() }), +})) + +vi.mock('axios', () => { + const mk = () => ({ + get: vi.fn(), post: vi.fn(), put: vi.fn(), delete: vi.fn(), + interceptors: { request: { use: vi.fn() }, response: { use: vi.fn() } }, + defaults: { headers: { common: {} } }, + }) + return { + default: Object.assign( + { get: vi.fn(), post: vi.fn(), put: vi.fn(), delete: vi.fn(), create: mk }, + { interceptors: { request: { use: vi.fn() }, response: { use: vi.fn() } }, + defaults: { headers: { common: {} } } }, + ), + } +}) + +import axios from 'axios' +import { useClientPortalStore } from '@/stores/clientPortal' + +const ask = (id, over = {}) => ({ + id, agent_name: 'scout', kind: 'question', priority: 'medium', + title: 'Ship it?', question: 'Ship it?', options: ['yes', 'no'], + created_at: '2026-08-19T10:00:00Z', expires_at: null, status: 'pending', + chat_id: null, ...over, +}) + +let store + +beforeEach(() => { + localStorage.clear() + sessionStorage.clear() + setActivePinia(createPinia()) + vi.clearAllMocks() + localStorage.setItem('trinity.portalToken', 'portal-token') // a signed-in client + store = useClientPortalStore() +}) + +describe('one list, three renderings', () => { + it('the count, the agent filter and the full list all come from one fetch', async () => { + axios.get.mockResolvedValueOnce({ + data: [ask('a1'), ask('a2', { agent_name: 'sage' })], + }) + + await store.fetchAsks() + + expect(axios.get).toHaveBeenCalledTimes(1) + expect(store.askCount).toBe(2) // sidebar + expect(store.asksForAgent('scout').map((a) => a.id)).toEqual(['a1']) // agent page + expect(store.asks).toHaveLength(2) // inline / global + }) + + it('answering removes the row, so every surface clears with no second call', async () => { + axios.get.mockResolvedValueOnce({ data: [ask('a1'), ask('a2')] }) + await store.fetchAsks() + + axios.post.mockResolvedValueOnce({ data: { ...ask('a1'), status: 'responded' } }) + await store.answerAsk('a1', { response: 'yes' }) + + expect(store.askCount).toBe(1) + expect(store.asksForAgent('scout').map((a) => a.id)).toEqual(['a2']) + expect(axios.get).toHaveBeenCalledTimes(1) // no refetch needed + }) + + it('an expired ask still renders but is not counted', async () => { + // #1142 deletes terminal rows, so expiry must be VISIBLE while it exists — but + // it is not something to nag about: the person cannot answer it any more. + axios.get.mockResolvedValueOnce({ + data: [ask('a1'), ask('a2', { status: 'expired' })], + }) + + await store.fetchAsks() + + expect(store.asks).toHaveLength(2) + expect(store.askCount).toBe(1) + expect(store.asksForAgent('scout')).toHaveLength(2) + }) + + it('narrows server-side when an agent is named', async () => { + axios.get.mockResolvedValueOnce({ data: [ask('a1')] }) + await store.fetchAsks('scout') + + const [, config] = axios.get.mock.calls[0] + expect(config.params).toEqual({ agent_name: 'scout' }) + }) +}) + +describe('an unentitled build says nothing', () => { + it('404 leaves the feature unavailable and logs nothing', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + axios.get.mockRejectedValueOnce({ response: { status: 404 } }) + + await store.fetchAsks() + + expect(store.asksAvailable).toBe(false) + expect(store.asks).toEqual([]) + expect(warn).not.toHaveBeenCalled() + warn.mockRestore() + }) + + it('403 is equally silent — the module is present but not entitled', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + axios.get.mockRejectedValueOnce({ response: { status: 403 } }) + + await store.fetchAsks() + + expect(store.asksAvailable).toBe(false) + expect(warn).not.toHaveBeenCalled() + warn.mockRestore() + }) + + it('a real failure is reported once, and clears the list rather than showing stale asks', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + axios.get.mockResolvedValueOnce({ data: [ask('a1')] }) + await store.fetchAsks() + + axios.get.mockRejectedValueOnce({ response: { status: 500 } }) + await store.fetchAsks() + + expect(warn).toHaveBeenCalledTimes(1) + expect(store.asks).toEqual([]) + expect(store.asksAvailable).toBe(false) + warn.mockRestore() + }) + + it('does not poll at all when nobody is signed in', async () => { + localStorage.clear() + setActivePinia(createPinia()) + const anon = useClientPortalStore() + + await anon.fetchAsks() + + expect(axios.get).not.toHaveBeenCalled() + }) +}) + +describe('answering surfaces the backend refusal', () => { + it('propagates the error and keeps the ask, so nothing looks resolved that is not', async () => { + axios.get.mockResolvedValueOnce({ data: [ask('a1')] }) + await store.fetchAsks() + + axios.post.mockRejectedValueOnce({ + response: { status: 409, data: { detail: { code: 'expired', message: 'This ask expired before it was answered.' } } }, + }) + + await expect(store.answerAsk('a1', { response: 'yes' })).rejects.toBeTruthy() + expect(store.askCount).toBe(1) + }) +}) diff --git a/tests/unit/test_ent364_addressed_asks.py b/tests/unit/test_ent364_addressed_asks.py new file mode 100644 index 0000000000..554e5c1762 --- /dev/null +++ b/tests/unit/test_ent364_addressed_asks.py @@ -0,0 +1,144 @@ +"""`operator_queue.addressed_to_email` and its roster validation (ent#364). + +The OSS half of workspace asks: an edition-agnostic primitive (the column and the +rule that fills it), with the read/answer surface gated separately — the same split +as `users.suspended_at` in #995, where OSS owns the column AND its enforcement. + +This field is the one piece of an agent-authored queue item that is an +AUTHORIZATION decision: it decides who may answer the ask and whose Workspace +sidebar it appears in. Which is exactly why it is not a key inside `context` — +`context` is free-form agent JSON whose hygiene clamp bounds size and type only, so +an addressee living there would be an agent choosing its own audience. +""" +from __future__ import annotations + +import pytest + +pytestmark = pytest.mark.unit + + +@pytest.fixture() +def roster(monkeypatch): + state = {"members": {("scout", "client@example.com")}, "calls": [], "raises": False} + + import client_portal.service as portal_service + + def _on_roster(agent_name, email, include_owned=False): + state["calls"].append((agent_name, email, include_owned)) + if state["raises"]: + raise RuntimeError("roster down") + return (agent_name, (email or "").lower()) in state["members"] + + monkeypatch.setattr(portal_service, "agent_on_roster", _on_roster) + return state + + +def _clamp(raw_email, agent="scout"): + from services.operator_queue_service import _clamp_ingested_item + + out = _clamp_ingested_item( + {"id": "req-1", "title": "t", "question": "q", "addressed_to_email": raw_email}, + agent, + ) + return out["addressed_to_email"] + + +def test_an_on_roster_address_is_kept_and_normalised(roster): + assert _clamp("Client@Example.com ") == "client@example.com" + + +def test_an_off_roster_address_becomes_an_operator_ask(roster): + """The whole point: an agent may not decide who answers for it. Dropped to + NULL — which is a perfectly valid item, just one the operator answers.""" + assert _clamp("stranger@example.com") is None + + +def test_an_address_for_a_different_agent_does_not_carry_over(roster): + """Roster membership is per agent. `sage` cannot address `scout`'s client.""" + assert _clamp("client@example.com", agent="sage") is None + + +@pytest.mark.parametrize("bad", [None, 123, {"email": "x@y.z"}, [], "", " ", "not-an-email"]) +def test_malformed_values_are_dropped_without_a_roster_call(roster, bad): + """Refused before the DB read: a non-string or address-shaped-nothing can never + match a roster row, and a lookup per garbage value is a free amplification.""" + assert _clamp(bad) is None + assert roster["calls"] == [] + + +def test_an_absurdly_long_value_is_refused_before_the_lookup(roster): + from services.operator_queue_service import OPERATOR_QUEUE_EMAIL_MAX + + assert _clamp("a" * (OPERATOR_QUEUE_EMAIL_MAX + 1) + "@example.com") is None + assert roster["calls"] == [] + + +def test_a_roster_failure_fails_closed(roster): + """Addressing an ask we could not validate is worse than not addressing it.""" + roster["raises"] = True + assert _clamp("client@example.com") is None + + +def test_the_lookup_never_unlocks_the_owned_agents_branch(roster): + """`include_owned=True` would hand a client agents nobody shared with them — + see `agent_on_roster`'s own docstring. It is the rule here, not a default.""" + _clamp("client@example.com") + assert roster["calls"] == [("scout", "client@example.com", False)] + + +def test_the_addressee_is_never_read_from_context(roster): + """A `context.addressed_to_email` must not become the addressee: `context` is + agent-authored and only size/type-clamped, so honouring it would reintroduce + exactly the hole the column exists to close.""" + from services.operator_queue_service import _clamp_ingested_item + + out = _clamp_ingested_item( + { + "id": "req-2", "title": "t", "question": "q", + "context": {"addressed_to_email": "client@example.com"}, + }, + "scout", + ) + assert out["addressed_to_email"] is None + + +# --- persistence ------------------------------------------------------------ + +@pytest.fixture() +def queue_db(tmp_path, monkeypatch): + import sqlite3 + + db_file = tmp_path / "trinity-asks-oss.db" + monkeypatch.setenv("TRINITY_DB_PATH", str(db_file)) + + import db.connection as conn_mod + monkeypatch.setattr(conn_mod, "DB_PATH", str(db_file)) + + from db.schema import init_schema + raw = sqlite3.connect(db_file) + init_schema(raw.cursor(), raw) + raw.commit() + raw.close() + yield str(db_file) + + +def test_the_column_round_trips(queue_db, roster): + from database import db + + item_id = db.create_operator_queue_item("scout", { + "id": "req-round-trip", "title": "t", "question": "q", + "addressed_to_email": "client@example.com", + }) + + assert db.get_operator_queue_item(item_id)["addressed_to_email"] == "client@example.com" + + +def test_an_unaddressed_item_reads_back_as_none(queue_db, roster): + """Every pre-ent#364 row is this row: NULL means the operator answers.""" + from database import db + + item_id = db.create_operator_queue_item("scout", { + "id": "req-operator", "title": "t", "question": "q", + }) + + assert db.get_operator_queue_item(item_id)["addressed_to_email"] is None From 6903bff4d6d8d32b0b09fb80b6a1409201b51b14 Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Wed, 19 Aug 2026 13:11:13 +0300 Subject: [PATCH 2/3] fix(db): guard the 0039 add_column so a fresh PostgreSQL boot survives (ent#364) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 0001_baseline builds operator_queue from the live db/tables.py metadata, which now declares addressed_to_email — so on a FRESH PostgreSQL database the column already exists by the time 0039 runs, and the unguarded add_column aborted the Alembic upgrade with DuplicateColumn. That kills boot, not just the revision. The SQLite track hid this: _safe_add_column already guards there, so the whole unit suite stayed green. Found by booting the branch against a real Postgres; CI's pg-migrations job failed on it independently. Guarded with the same _has_column idiom 0031 uses. Co-Authored-By: Claude Opus 5 (1M context) --- .../0039_operator_queue_addressed_to.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/src/backend/migrations/versions/0039_operator_queue_addressed_to.py b/src/backend/migrations/versions/0039_operator_queue_addressed_to.py index fd90284eb7..df5e8712cd 100644 --- a/src/backend/migrations/versions/0039_operator_queue_addressed_to.py +++ b/src/backend/migrations/versions/0039_operator_queue_addressed_to.py @@ -25,9 +25,24 @@ depends_on = None +def _has_column(bind, table: str, column: str) -> bool: + return column in {c["name"] for c in sa.inspect(bind).get_columns(table)} + + def upgrade() -> None: - op.add_column("operator_queue", sa.Column("addressed_to_email", sa.Text(), nullable=True)) + # Guarded because a FRESH PostgreSQL database already carries this column: + # 0001_baseline builds the table from the live `db/tables.py` metadata, which + # declares it. Without the guard the revision aborts boot with DuplicateColumn + # on every new install. The SQLite track hides this — `_safe_add_column` guards + # for us there — so the guard has to be explicit here (0031 does the same). + bind = op.get_bind() + if not _has_column(bind, "operator_queue", "addressed_to_email"): + op.add_column( + "operator_queue", sa.Column("addressed_to_email", sa.Text(), nullable=True) + ) def downgrade() -> None: - op.drop_column("operator_queue", "addressed_to_email") + bind = op.get_bind() + if _has_column(bind, "operator_queue", "addressed_to_email"): + op.drop_column("operator_queue", "addressed_to_email") From c9c545e8929f9b7c4d20fcc1507a38f59b1c9509 Mon Sep 17 00:00:00 2001 From: trinity-ability Date: Wed, 19 Aug 2026 12:24:42 +0100 Subject: [PATCH 3/3] chore(config): wire OPERATOR_QUEUE_EMAIL_MAX into both compose files + .env.example (ent#364) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The new addressee-email clamp read in operator_queue_service.py was env-tunable in code but unreachable from .env — its siblings (OPERATOR_QUEUE_TITLE_MAX, OPERATOR_QUEUE_EXECUTION_ID_MAX) are wired per the #1632 convention, so wire it the same way. Default unchanged (320, RFC 5321). Co-Authored-By: Claude Fable 5 --- .env.example | 1 + docker-compose.prod.yml | 1 + docker-compose.yml | 1 + 3 files changed, 3 insertions(+) diff --git a/.env.example b/.env.example index 7324c0cdbf..7fe8680f52 100644 --- a/.env.example +++ b/.env.example @@ -315,6 +315,7 @@ OPERATOR_QUEUE_CONTEXT_MAX_BYTES=8192 # context serialized-byte cap (mar OPERATOR_QUEUE_OPTIONS_MAX_BYTES=4096 # options serialized-byte cap (dropped-marker over) OPERATOR_QUEUE_ID_MAX=256 # max agent-supplied id length OPERATOR_QUEUE_EXECUTION_ID_MAX=128 # max execution_id preserved in a context marker +OPERATOR_QUEUE_EMAIL_MAX=320 # ent#364: addressee email length clamp (RFC 5321) OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS=300 # one flood alert per agent per this window OPERATOR_ALERT_MAX_PENDING_PER_TYPE=5 # per-(agent, type) pending budget for agent-influenceable platform alerts (#1677) diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index eeb4c6c02a..5c6a8d3bc2 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -156,6 +156,7 @@ services: - OPERATOR_QUEUE_OPTIONS_MAX_BYTES=${OPERATOR_QUEUE_OPTIONS_MAX_BYTES:-4096} # options byte cap (dropped over) - OPERATOR_QUEUE_ID_MAX=${OPERATOR_QUEUE_ID_MAX:-256} # max agent-supplied id length - OPERATOR_QUEUE_EXECUTION_ID_MAX=${OPERATOR_QUEUE_EXECUTION_ID_MAX:-128} # max execution_id kept in a marker + - OPERATOR_QUEUE_EMAIL_MAX=${OPERATOR_QUEUE_EMAIL_MAX:-320} # ent#364: addressee email length clamp (RFC 5321) - OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS=${OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS:-300} # one flood alert / agent / window - OPERATOR_ALERT_MAX_PENDING_PER_TYPE=${OPERATOR_ALERT_MAX_PENDING_PER_TYPE:-5} # per-(agent,type) platform-alert budget (#1677) - GITHUB_PAT=${GITHUB_PAT} diff --git a/docker-compose.yml b/docker-compose.yml index 870eb40cab..3df878d8cd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -121,6 +121,7 @@ services: - OPERATOR_QUEUE_OPTIONS_MAX_BYTES=${OPERATOR_QUEUE_OPTIONS_MAX_BYTES:-4096} # options byte cap (dropped over) - OPERATOR_QUEUE_ID_MAX=${OPERATOR_QUEUE_ID_MAX:-256} # max agent-supplied id length - OPERATOR_QUEUE_EXECUTION_ID_MAX=${OPERATOR_QUEUE_EXECUTION_ID_MAX:-128} # max execution_id kept in a marker + - OPERATOR_QUEUE_EMAIL_MAX=${OPERATOR_QUEUE_EMAIL_MAX:-320} # ent#364: addressee email length clamp (RFC 5321) - OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS=${OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS:-300} # one flood alert / agent / window - OPERATOR_ALERT_MAX_PENDING_PER_TYPE=${OPERATOR_ALERT_MAX_PENDING_PER_TYPE:-5} # per-(agent,type) platform-alert budget (#1677) - GITHUB_PAT=${GITHUB_PAT:-}