Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
1 change: 1 addition & 0 deletions docker-compose.prod.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down
1 change: 1 addition & 0 deletions docker-compose.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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:-}
Expand Down
3 changes: 3 additions & 0 deletions docs/memory/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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);
Expand Down
24 changes: 24 additions & 0 deletions src/backend/db/migrations.py
Original file line number Diff line number Diff line change
Expand Up @@ -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).

Expand Down Expand Up @@ -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),
]
7 changes: 7 additions & 0 deletions src/backend/db/operator_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions src/backend/db/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
)
""",
Expand Down
4 changes: 4 additions & 0 deletions src/backend/db/tables.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
"""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 _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:
# 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:
bind = op.get_bind()
if _has_column(bind, "operator_queue", "addressed_to_email"):
op.drop_column("operator_queue", "addressed_to_email")
54 changes: 52 additions & 2 deletions src/backend/services/operator_queue_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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)
Expand Down
7 changes: 7 additions & 0 deletions src/frontend/src/components/portal/PortalAgentPage.vue
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,12 @@
>Start a chat</button>
</div>

<!-- ent#364: asks this agent raised for you, at the top of its page where a
pending decision belongs. Renders nothing when there are none. -->
<div v-if="store.asksForAgent(agentName).length" class="px-3 sm:px-6 pt-3">
<PortalAsks :agent-name="agentName" />
</div>

<!-- Stats strip -->
<div class="px-3 sm:px-6 pt-4">
<div class="flex flex-wrap items-end gap-x-8 gap-y-3">
Expand Down Expand Up @@ -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'

Expand Down
Loading
Loading