diff --git a/README.md b/README.md index 776bbf2..7bbe24f 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,14 @@ MCP authorization): 24 slices, each merged with a merge commit and tagged `slice protected server's `401` with "authorize" on the `/mcp` page and keeps the token it obtains. Trinity issues no production authority: the personal issuer refuses to start under an external authority adapter, by construction (`docs/07-security-model.md`). +- **Reachable from elsewhere.** A gateway is a channel Trinity answers from: the adapter carries + text, and everything else (the session, the gate, the receipts) is the same machinery the + desktop uses. A sender Trinity does not know gets a pairing code shown on the `/gateways` page + and nothing else, no session and no model; once paired they get slash commands, the answer + streaming back as the turn runs, and any approval the turn raises rendered into the same + conversation. What a channel may approve is capped below what the desktop may: a `write` at + most by default, so an `exec` or a `destructive` request is decided at the machine, and the + refusal is receipted. The console adapter ships (`mix trinity.console`); the platforms are next. - **Runs on a schedule.** Tasks on the `/tasks` page: a prompt, a persona, the skills to hint, and when (a cron expression, a one-shot time, or a phrase like "every weekday at 9am" the model turns into cron). Each run is a fresh conversation you can open, its result waits on @@ -65,7 +73,7 @@ MCP authorization): 24 slices, each merged with a merge commit and tagged `slice curator that marks old memories stale and archives the untouched ones (never deleting) run on the same queues; `/oban` shows the jobs. -Not there yet: messaging gateways and subagents (M5b), +Not there yet: the messaging platforms themselves and subagents (M5b), the native desktop shell and signed releases (M6), executable skills in a sandbox (M7). `ROADMAP.md` carries the live status of every slice, and the [Milestones](#milestones) section below explains how to read it. @@ -107,7 +115,7 @@ model registry is `config/llm.exs`; keys come from the environment (`.env` is gi pages: `/` conversations, `/s/:id` a conversation and `/s/:id/receipts` its receipts, `/search` full-text search, `/personas` and `/memory` the persona and its memory (with the semantic tab and the embedding model's download), `/skills` the skills, the changes waiting for your decision and -the learn form, `/mcp` the MCP servers you connect to (their health and the tools they +the learn form, `/gateways` the channels you can reach Trinity from, `/mcp` the MCP servers you connect to (their health and the tools they contribute), `/permissions` the rules and pending approvals (a server's question to you, when one of its tools asks for input mid-call, is answered there too), `/tasks` the scheduled tasks and their results, `/oban` the jobs (in development, or when configured), `/settings` the export. `mix trinity.export` and `mix trinity.import` do what `/settings` does from a terminal; diff --git a/ROADMAP.md b/ROADMAP.md index a30926b..0b889bd 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -56,7 +56,7 @@ standards register names the rows that ask for them. | 060 | MCP client: Trinity's thin driver (2026-07-28 preferred, 2025-11-25 compat, MRTR, Tasks) | 6 MCP | L | 059, 021 | approved | | 061 | MCP server (stateless 2026-07-28 + compat, MRTR approvals, headless profile) | 6 MCP | M | 060, 024 | approved | | 062 | MCP authorization: OAuth client role, resource server, personal-profile authorization server (EMA removed 2026-09-22) | 6 MCP | L | 061 | approved | -| 070 | Gateway core: adapter behaviour, routing, PubSub fan-out | 7 Gateways | M | 012 | planned | +| 070 | Gateway core: adapter behaviour, routing, PubSub fan-out | 7 Gateways | M | 012 | done | | 071 | Gateway: Telegram | 7 Gateways | S | 070 | planned | | 072 | Gateway: Discord (Nostrum) | 7 Gateways | S | 070 | planned | | 080 | Subagents + delegation | 8 Orchestration | M | 020, 023 | planned | diff --git a/config/config.exs b/config/config.exs index 99b5214..dcb2c8d 100644 --- a/config/config.exs +++ b/config/config.exs @@ -214,3 +214,7 @@ config :phoenix, :json_library, Jason # Import environment specific config. This must remain at the bottom # of this file so it overrides the configuration defined above. import_config "#{config_env()}.exs" + +# Slice 070: a scheduled run can be delivered to a gateway conversation; `deliver_to`'s "kind" +# names the implementation (slice 050's `Trinity.Scheduler.Delivery.for/1`). +config :trinity, :deliveries, %{"gateway" => Trinity.Scheduler.Delivery.Gateway} diff --git a/docs/01-architecture.md b/docs/01-architecture.md index 123f21c..0b29e83 100644 --- a/docs/01-architecture.md +++ b/docs/01-architecture.md @@ -53,7 +53,8 @@ Trinity.Application │ # a dead one dropped), writes the index rows on first read ├── Oban # cron + durable jobs. Slice 050 ├── Trinity.MCP.Supervisor # MCP clients (one per server, lib per 059) + server. Slice 059-062 -├── Trinity.Gateways.Supervisor # adapters (Telegram, Discord…). Slice 070+ +├── Trinity.Gateways.Router # one road from a channel to a session. Slice 070 +├── Trinity.Gateways.Console # the in-process adapter; platforms at 071, 072 ├── Trinity.Subagents.Supervisor (DynamicSupervisor) # Slice 080 ├── Trinity.Sandbox.Supervisor # Luerl workers. Slice 110 ├── Trinity.Desktop # ex_tauri bridge (tray, notifications). Slice 100 @@ -91,7 +92,7 @@ without anything failing. | `Trinity.Skills` | SKILL.md parsing, registry, loader, manager, scanner (as built at 040: parser, sources, registry, index, the three tools; at 041: staging, promotion, manager, scanner, diff, learn, `skill_manage` and `learn`) | Repo, Permissions, **Effects**, **Receipts**, Sandbox (as built at 041: Tools, Memory, Permissions, Receipts and LLM; Effects is not a dependency: the promotion is not a tool call, it writes its own effect receipt; Tools never depends on Skills) | | `Trinity.Scheduler` | Oban workers for agent tasks, delivery (as built at 050: the `tasks` and `task_runs` rows, `next_run_at` computed from a cron expression or a one-shot datetime, `Workers.Tick` (the one Cron plugin entry, every minute, enqueuing `Workers.RunTask` for what is due), `RunTask` (one turn in a fresh `origin: "cron"` session, the summary, the delivery), `Delivery` with `Delivery.Desktop`, `Parse.human/2` (a phrase to cron through the model, checked by Oban's parser); Oban itself on the app's repo, Lite on SQLite and Basic on Postgres, queues `agent_tasks`, `memory`, `maintenance`; the observer as `Trinity.Memory.ObserverWorker` and the curator as `Trinity.Memory.Curator` live in Memory) | Sessions, LLM, **Repo** (Gateways at 070) | | `Trinity.MCP` | Client manager, tool bridge, server (as built at 059: the boundary alone, holding the core's version, its JSON depth and its telemetry event names; at 060: `Servers` and `ServerConfig` (the rows), `Supervisor` and `Boot` (one `Client` per enabled row), `Client` with its `Wire` (the outbound request, the headers, the core's decoder and validator) and two transports (`Transport.Stdio`, a child on a Port; `Transport.HTTP`, one POST per request), `Bridge` (the one tool module every MCP tool runs through) and `Client.Auth` (062's seam, a static token at 060); at 061: `Server` (the module above the core, `@behaviour BeamMCP.Server`, handed to both transports through `:server`; `tools/call` answered through the membrane, everything else the core's), `Server.Catalog` (the exported tools, sorted), `Server.Exports` (the configured names, `:catalog` refused), `Server.Session` (the `origin: "mcp"` session with its own persona), `Server.Envelope` and `Server.Replay` (the sealed `requestState` and the nonce table), `Server.Auth.Local` (the static bearer, 062's seam on the server side), `Server.Plug` (the `POST /mcp` endpoint, mounted in `TrinityWeb.Endpoint` ahead of the parsers) and `Server.Stdio` with `mix trinity.mcp.stdio`; at 062: `Trinity.MCP.Auth`, a boundary of its own under `Trinity.MCP` with `deps: []` (the behaviour `authorize/2` answering a `Principal`, never the token; `Config`, `Token`, `JWKS`, `Discovery`, `Scopes`; the profiles `Local`, `External` and `Embedded` with `Embedded.Keys`; the client role `Client` with `Client.Store`), and `AuthHost` in `Trinity.MCP` (the configuration from the tree's paths and the authority selection, the personal profile's boot, the refusal receipts, the client role's flow for the page); `Server.Auth.Local` is gone, `Server.Plug` authorizes ahead of the transport and answers `401` with a challenge, `Server` checks scope before the gate) | Tools, **Effects**, **Permissions**, Memory (as built at 059: a top-level boundary, like `Trinity.Smoke`, with `deps: [Trinity, BeamMCP.JSON]`; the boundary compiler checks every call into the `beam_mcp` application and this boundary alone lists its modules; at 060 the deps are `[Trinity, BeamMCP.JSON, BeamMCP.Schema]`, Trinity's tools, permissions and receipts reached through `Trinity`'s exports, and `Trinity.Application` and `TrinityWeb` list `Trinity.MCP`; at 061 the core's `Server`, `Catalog`, `ToolSpec` and both transports are listed too, and `Trinity` exports `Tools.Registry` and `Effects.Runner` to it) | -| `Trinity.Gateways` | Adapter behaviour, router, allowlists, pairing | Sessions, **Permissions**, PubSub | +| `Trinity.Gateways` | Adapter behaviour, router, allowlists, pairing (as built at 070: `Adapter` (the behaviour: `child_spec/1`, `capabilities/0`, `deliver/2`, `format/2`, `render_approval/2`, plus `name/1` deriving the row's adapter name from the module), `Console` (the in-process adapter the tests and `mix trinity.console` talk to), `Format` (chunking to the channel's limit, re-opening a split code fence, markdown reduced to text where a channel renders none, the default approval rendering), `Identity` and `Identities` (the `gateway_identities` rows, the pairing code and the allowlist), `Router` (one process: admission, then the token bucket, then commands, then the session; it subscribes to `session:` and `approvals:` per conversation and coalesces deltas into an edited message), `Commands` (the slash registry, including `/attach`, `/approve` and `/deny`), `Cap` (the channel trust cap) and `Receipts` (the capped-decision receipt)) | Sessions, **Permissions**, PubSub | | `Trinity.Subagents` | Delegation, result collection | Sessions, Tools | | `Trinity.Sandbox` | Luerl runners, resource limits | none | | `Trinity.Desktop` | ex_tauri bridge | PubSub | diff --git a/docs/05-data-model.md b/docs/05-data-model.md index 1214e51..7fc3790 100644 --- a/docs/05-data-model.md +++ b/docs/05-data-model.md @@ -173,7 +173,7 @@ One row per request, the audit trail this slice owns; slice 024 reads it for dec | prompt | text | the user message of the run's turn | | persona_id | fk personas, nullable | the default persona when unset | | skill_names | {array, string} | hinted to the model in the prompt | -| deliver_to | map | `%{"kind" => "desktop"}` at 050; gateways name their kind at 070 | +| deliver_to | map | `%{"kind" => "desktop"}` at 050; at 070 `%{"kind" => "gateway", "adapter" => "console", "conversation" => "c-1"}` names a channel | | enabled | boolean | the tick enqueues enabled tasks only; a `once` task is disabled once enqueued | | timeout_ms | integer | one run's bound (600 000 by default) | | last_run_at, next_run_at | timestamp | `next_run_at` is computed by `Trinity.Scheduler` from the schedule, not mirrored from Oban: the tick (one Cron plugin entry a minute) enqueues what is due and advances it | @@ -194,6 +194,26 @@ Oban's own `oban_jobs` (and `oban_peers`) carry the jobs; the pruner keeps a wee The curator (slice 050) adds `stale_at` and `archived_at` to `memories` (docs above): stale is still recalled and shown as such; archived leaves recall and stays in the row. +### gateway_identities (Slice 070, as built) +Who outside this machine may talk to Trinity. The key is the pair, because an external id means +nothing outside the platform that issued it. + +| column | type | notes | +|---|---|---| +| adapter | string | the adapter's own name (`Trinity.Gateways.Adapter.name/1`: "console"), half the key | +| external_user_id | string | the platform's id for the person; unique with `adapter` | +| display_name | string, nullable | whatever the platform offered, for the page | +| state | string | "pending" (a code was shown) \| "paired" \| "revoked"; a revoked row is kept, so turning someone away is recorded rather than forgotten | +| code | string, nullable | the six-character pairing code, shown on `/gateways` and nowhere else; cleared once spent | +| code_expires_at | timestamp, nullable | ten minutes; a pending sender who comes back later is shown a new code rather than locked out | +| paired_at, revoked_at | timestamp, nullable | | +| last_conversation | string, nullable | where the pairing was asked from, so the answer goes back there | + +A gateway session is an ordinary session with `origin` = the adapter's name and `origin_ref` +`%{"adapter", "conversation", "external_user_id"}` (010's columns; this slice adds no column to +`sessions`). A scheduled task delivers to one with +`deliver_to: %{"kind" => "gateway", "adapter" => …, "conversation" => …}`. + ### usage_events (Slice 011; the ledger and budgets that read it are Slice 090) One row per completed call, as built at slice 011: diff --git a/docs/07-security-model.md b/docs/07-security-model.md index 5a7399b..5b70545 100644 --- a/docs/07-security-model.md +++ b/docs/07-security-model.md @@ -282,17 +282,39 @@ regulated effect. - Never in DB, never in logs, never in prompts. `mix gate` runs a regex secret scan on the diff. - Provider keys are read at call time by the provider module; not held in Session state. -## Gateways - -- Allowlist of external user ids; DM pairing code flow (Slice 070). Unknown senders get a pairing prompt, nothing else. -- Rate limits per external user. +## Gateways (Slice 070, as built) + +- **An unknown sender gets a pairing prompt and nothing else.** No session is created, no model is + called and nothing of the message is read but the code it might be: `Trinity.Gateways.Router` + asks `Identities.admit/3` before anything else happens. The code is six characters from an + alphabet without the pairs a person misreads, lives ten minutes, and is compared in constant + time. It is shown on `/gateways` and nowhere else, which is the whole of the proof: a sender + who can read it is at the owner's desktop. A configured allowlist + (`config :trinity, :gateways, allowlist: [{adapter, id}]`) pairs a known id at first sight. +- **An identity is `(adapter, external_user_id)`**, because an id means nothing outside the + platform that issued it, and a revoked row is kept rather than deleted: turning someone away is + a thing the page records rather than a thing it forgets. +- **Rate limits per external user**: a token bucket in the router's own state, above the session, + so a flood costs a row lookup and nothing more. A throttled sender is told, not dropped. +- **An adapter carries text and nothing else.** It never calls the LLM, never touches a session + and never decides an approval; that rule is what lets a platform be added as one module and a + configuration entry. A command that raises costs its own message: the router holds every + conversation's binding and rescues rather than dying with them. - **Channel trust cap.** The allowlist, pairing and rate limits control *who may talk to the bot*. They do not control *what an approval arriving from that channel may authorise*, and an approval surface is exactly as trustworthy as the account behind it. So each channel carries a cap on the risk tier it can approve, and the cap is applied after the gate's own decision, never instead of it. Default: `:read`, `:network` and `:write` - are approvable from any paired gateway; `:exec` and `:destructive` are **desktop or console only**. The cap is - configurable and the default is the safe one. A capped request is not silently dropped: the requester is told - the decision must be made on the desktop, and the refusal is receipted like any other. + are approvable from any paired gateway; `:exec` and `:destructive` are **desktop only**. The cap is + configurable per adapter (`config :trinity, :gateways, caps: %{"console" => :write}`) and the default is the + safe one. A capped request is not silently dropped: the requester is told the decision must be made on the + desktop, and the refusal is receipted like any other (`Trinity.Gateways.Receipts`, basis `channel_cap`, on the + session's own chain). + + **Amended at 070**: this paragraph read "desktop or console only" when it was written, before there was a + console adapter. There is one now, and it is capped like every other channel rather than exempted: an + in-process adapter is still a channel, and making it the one that may approve a destructive effect would be a + hole shaped exactly like the thing the cap exists for. `Trinity.Gateways.Cap` applies the ceiling to every + adapter, `Console` included, and an unknown tier is refused rather than waved through. ## Effects and receipts (Slice 024, as built) diff --git a/lib/mix/tasks/trinity.console.ex b/lib/mix/tasks/trinity.console.ex new file mode 100644 index 0000000..2126922 --- /dev/null +++ b/lib/mix/tasks/trinity.console.ex @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Mix.Tasks.Trinity.Console do + @shortdoc "Talks to Trinity from this terminal through the console gateway" + @moduledoc """ + Slice 070. Starts the application and talks to it through `Trinity.Gateways.Console`, the + in-process adapter, so the gateway layer can be used and watched without a platform account: + + mix trinity.console + + It is a real channel and not a back door. The first message is answered with a pairing code, + as any unknown sender's is; the code is shown on the desktop's `/gateways` page, and typing it + here pairs this terminal. Every slash command works (`/help` lists them), an approval raised by + a turn appears here, and the channel's tier ceiling applies to `/approve` exactly as it would + on a messaging platform: an `exec` or `destructive` request is decided on the desktop. + + Two things to know, as with `mix trinity.mcp.stdio`: this is a whole Trinity and the data + directory admits one at a time (`Trinity.DataDir.Lock`), so run it when the desktop does not, + or point it at its own directory with `TRINITY_DATA_DIR`. Ctrl-C twice leaves. + """ + use Boundary, classify_to: Trinity.Gateways + use Mix.Task + + alias Trinity.Gateways.{Console, Router} + + @conversation "terminal" + + @impl Mix.Task + def run(argv) do + Mix.Task.run("app.start") + user = Enum.at(argv, 0, System.get_env("USER") || "owner") + {:ok, _} = ensure_started(Console) + {:ok, _} = ensure_started(Router) + + IO.puts("Trinity console. The first message will be answered with a pairing code.") + IO.puts("Say /help for the commands, or Ctrl-C twice to leave.\n") + loop(user) + end + + defp ensure_started(module) do + case module.start_link([]) do + {:ok, pid} -> {:ok, pid} + {:error, {:already_started, pid}} -> {:ok, pid} + other -> other + end + end + + defp loop(user) do + case IO.gets("> ") do + :eof -> + :ok + + {:error, reason} -> + Mix.shell().error("console: #{inspect(reason)}") + + line -> + line |> String.trim() |> send_line(user) + loop(user) + end + end + + defp send_line("", _user), do: :ok + + defp send_line(text, user) do + before = length(Console.delivered(@conversation)) + _ = Router.inbound(Console, @conversation, user, text) + # The reply streams, so the terminal waits a moment for what the channel was shown rather + # than printing nothing and looking broken. + print_new(before, 0) + end + + defp print_new(before, waited) when waited < 30_000 do + delivered = Console.delivered(@conversation) + + if length(delivered) > before do + Process.sleep(200) + Console.text(@conversation) |> Enum.drop(visible_before(before)) |> Enum.each(&IO.puts/1) + else + Process.sleep(100) + print_new(before, waited + 100) + end + end + + defp print_new(_before, _waited), do: IO.puts("(no answer yet)") + + # How many visible messages the log held before this turn: an edit replaces a message rather + # than adding one, so counting the log is not counting what a person has seen. + defp visible_before(before) do + @conversation + |> Console.delivered() + |> Enum.take(before) + |> Enum.count(&match?({:message, _}, &1)) + end +end diff --git a/lib/trinity.ex b/lib/trinity.ex index 98ff881..f41399a 100644 --- a/lib/trinity.ex +++ b/lib/trinity.ex @@ -70,6 +70,16 @@ defmodule Trinity do Skills.Manager, Skills.Learn, # Slice 050: the tasks page reads and writes the scheduler. + # Slice 070: the /gateways page reads the channels, the identities and the cap; the + # scheduler's gateway delivery reaches the adapter through these too. `Gateways` is a + # boundary of its own under this one, as `Scheduler` is. + Gateways, + Gateways.Adapter, + Gateways.Cap, + Gateways.Console, + Gateways.Identities, + Gateways.Identity, + Gateways.Router, Scheduler, Scheduler.Task, Scheduler.Run, diff --git a/lib/trinity/gateways.ex b/lib/trinity/gateways.ex new file mode 100644 index 0000000..10fcd82 --- /dev/null +++ b/lib/trinity/gateways.ex @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways do + @moduledoc """ + Trinity reached from somewhere other than the desktop (slice 070): a chat channel, a terminal, + later a messaging platform. A gateway is an adapter process plus `Router`, and the rule that + makes the layer safe to extend is that an adapter carries text and nothing else: it never calls + the LLM, never touches a session, and never decides an approval. Whether an effect happens is + the permission gate's answer, as it is for the desktop, and what an approval arriving from a + channel may authorise is capped besides (`Cap`, docs/07 "Gateways"). + + Adapters are configured, not compiled in: `config :trinity, :gateways, adapters: [Module, …]`. + `Console` ships here and is what the tests and `mix trinity.console` talk to. + """ + use Boundary, + deps: [Trinity, Trinity.Sessions], + exports: [Adapter, Cap, Console, Format, Identities, Identity, Router] + + @doc "The adapter modules in force." + @spec adapters() :: [module()] + def adapters, do: Application.get_env(:trinity, :gateways, []) |> Keyword.get(:adapters, []) +end diff --git a/lib/trinity/gateways/adapter.ex b/lib/trinity/gateways/adapter.ex new file mode 100644 index 0000000..c12295d --- /dev/null +++ b/lib/trinity/gateways/adapter.ex @@ -0,0 +1,81 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Adapter do + @moduledoc """ + What a gateway is (slice 070): a process that carries messages between one outside platform and + `Trinity.Gateways.Router`, and nothing else. An adapter never calls the LLM, never touches + `Trinity.Sessions` and never decides an approval; it receives text from a platform, hands it to + the router, and renders back what the router gives it. That rule is what lets a platform be + added by writing one module and a configuration entry (docs/03, the modularity rule). + + `capabilities/0` is how the router talks to an adapter without knowing the platform: what the + channel can render, how long a message may be, and whether an already-sent message can be + edited (which is what decides whether a stream is delivered as edits or as one final message). + """ + + alias Trinity.Permissions.Approval + + @typedoc "The platform's own identifier for a conversation (a channel, a chat, a DM)." + @type conversation :: String.t() + + @typedoc """ + What a channel can carry. `max_length` is in graphemes, not bytes: a platform counts characters + and so does `Format.chunk/2`. `edits` says an adapter can replace a message it already sent, + which is how a stream reaches a channel without one message per delta. + """ + @type capabilities :: %{ + markdown: boolean(), + images: boolean(), + buttons: boolean(), + edits: boolean(), + max_length: pos_integer() + } + + @typedoc """ + What the router asks an adapter to do. `{:edit, ref, text}` carries the reference the adapter + returned when it sent the message being replaced; an adapter without `edits` never receives one. + """ + @type outbound :: + {:message, String.t()} + | {:edit, reference_id(), String.t()} + | {:typing, boolean()} + + @typedoc "Whatever the adapter uses to name a message it sent, opaque to the router." + @type reference_id :: term() + + @doc """ + The adapter's name as rows and receipts carry it: the last segment of its module, underscored, + so `Trinity.Gateways.Console` is `"console"`. It is derived and not declared, because a name a + module can contradict is a second source of truth (CLAUDE.md section 8: a name is a claim). + """ + @spec name(module()) :: String.t() + def name(adapter) when is_atom(adapter) do + adapter |> Module.split() |> List.last() |> Macro.underscore() + end + + @doc "The adapter's child specification; the gateway supervisor starts it with its configuration." + @callback child_spec(keyword()) :: Supervisor.child_spec() + + @doc "What this channel can carry." + @callback capabilities() :: capabilities() + + @doc """ + Delivers one outbound instruction to a conversation. A `{:message, _}` answers with a reference + the router can later edit, when the channel supports editing; anything else answers `:ok`. + """ + @callback deliver(conversation(), outbound()) :: :ok | {:ok, reference_id()} | {:error, term()} + + @doc """ + An assistant message as this platform's text, already split to `max_length`. The default is + `Trinity.Gateways.Format.format/2`, which is what an adapter should delegate to unless its + platform needs its own dialect. + """ + @callback format(String.t(), capabilities()) :: [String.t()] + + @doc """ + An approval request as something the conversation can answer. The default is + `Trinity.Gateways.Format.render_approval/2`, which writes the two commands out, because a + channel without buttons still has to be able to answer. + """ + @callback render_approval(Approval.t(), capabilities()) :: outbound() +end diff --git a/lib/trinity/gateways/cap.ex b/lib/trinity/gateways/cap.ex new file mode 100644 index 0000000..dede483 --- /dev/null +++ b/lib/trinity/gateways/cap.ex @@ -0,0 +1,83 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Cap do + @moduledoc """ + What an approval arriving from a channel may authorise (slice 070; docs/07 "Gateways", the + channel trust cap). + + The allowlist, the pairing and the rate limit control *who may talk to Trinity*. They say + nothing about *what an approval from that channel may allow to happen*, and an approval surface + is exactly as trustworthy as the account behind it: a messaging account is a password reset and + a phone number away from someone else's hands, and the desktop is not. So a tier ceiling is + applied to every decision arriving through a gateway, **after** the permission gate has made its + own decision and never instead of it: the gate can still refuse what the cap would have allowed. + + The default ceiling is `:write`, so `:read`, `:network` and `:write` are answerable from a + paired channel and `:exec` and `:destructive` are not. A capped request is answered, not + dropped: the conversation is told where the decision has to be made, and the refusal is + receipted like any other, because a silent refusal teaches a person that the bot is broken. + + The ceiling is per adapter (`config :trinity, :gateways, caps: %{"console" => :write}`) so a + deployment can raise or lower one channel without touching the others. `Console` is capped like + any other adapter and not exempted: docs/07 was written before this slice and reads "desktop or + console only", meaning the machine's own surfaces; an in-process adapter is still a channel, and + making it the one channel that may approve a destructive effect would be a hole shaped exactly + like the thing the cap exists for. + """ + + alias Trinity.Gateways.Adapter + + @order [:read, :network, :write, :exec, :destructive] + @default_ceiling :write + + @doc "The tiers, weakest first." + @spec order() :: [atom()] + def order, do: @order + + @doc "The ceiling in force for an adapter." + @spec ceiling(module() | String.t()) :: atom() + def ceiling(adapter) when is_atom(adapter) and not is_nil(adapter), + do: adapter |> Adapter.name() |> ceiling() + + def ceiling(adapter_name) when is_binary(adapter_name) do + :trinity + |> Application.get_env(:gateways, []) + |> Keyword.get(:caps, %{}) + |> Map.get(adapter_name, @default_ceiling) + end + + @doc """ + Whether a decision on a request of this tier may be made from this channel. An unknown tier is + refused: a tier this module does not know is not a tier it may wave through. + """ + @spec allows?(module() | String.t(), atom() | String.t() | nil) :: boolean() + def allows?(adapter, tier) do + with {:ok, tier} <- normalise(tier), + {:ok, ceiling} <- normalise(ceiling(adapter)) do + index(tier) <= index(ceiling) + else + :error -> false + end + end + + @doc "Why a request was capped, in words a person in the channel can act on." + @spec refusal(module() | String.t(), atom() | String.t() | nil) :: String.t() + def refusal(adapter, tier) do + "That is a #{tier} request, and this channel may approve up to #{ceiling(adapter)}. " <> + "Decide it on the desktop, on the permissions page." + end + + defp normalise(tier) when is_atom(tier) and not is_nil(tier), + do: normalise(Atom.to_string(tier)) + + defp normalise(tier) when is_binary(tier) do + case Enum.find(@order, &(Atom.to_string(&1) == tier)) do + nil -> :error + found -> {:ok, found} + end + end + + defp normalise(_other), do: :error + + defp index(tier), do: Enum.find_index(@order, &(&1 == tier)) +end diff --git a/lib/trinity/gateways/commands.ex b/lib/trinity/gateways/commands.ex new file mode 100644 index 0000000..faa4233 --- /dev/null +++ b/lib/trinity/gateways/commands.ex @@ -0,0 +1,174 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Commands do + @moduledoc """ + What a channel can ask for without talking to the model (slice 070): a registry of slash + commands, so a platform adapter adds none of its own and a new command is a clause here rather + than a change to the router. + + Every command answers with text the router sends back. A command never decides an approval by + itself: `/approve` and `/deny` go through + `Trinity.Permissions`, after `Trinity.Gateways.Cap` has said whether a channel of this kind may + answer a request of that tier at all (docs/07, the channel trust cap). + """ + + alias Trinity.Gateways.{Adapter, Cap, Format, Receipts, Router} + alias Trinity.Permissions + alias Trinity.Sessions + + @doc "The commands, with one line of help each." + @spec help() :: [{String.t(), String.t()}] + def help do + [ + {"/new", "start a fresh session for this conversation"}, + {"/sessions", "list the most recent sessions"}, + {"/attach ", "watch an existing session from here"}, + {"/approve ", "approve a pending request, up to this channel's tier"}, + {"/deny ", "deny a pending request"}, + {"/help", "this list"} + ] + end + + @doc "Runs a command for an inbound message: the text for the channel, and the router's state." + @spec handle(map(), map()) :: {String.t(), map()} + def handle(%{text: text} = message, state) do + {command, argument} = split(text) + run(command, argument, message, state) + end + + defp split(text) do + case text |> String.trim() |> String.split(" ", parts: 2) do + [command] -> {command, ""} + [command, rest] -> {command, String.trim(rest)} + end + end + + defp run("/help", _argument, _message, state) do + body = Enum.map_join(help(), "\n", fn {name, line} -> "#{name} — #{line}" end) + {"What I answer here:\n" <> body, state} + end + + defp run("/new", _argument, message, state) do + {"Started a new session.", forget(state, message)} + end + + defp run("/sessions", _argument, _message, state) do + case Sessions.list_sessions(limit: 5) do + [] -> + {"No sessions yet.", state} + + sessions -> + body = + Enum.map_join(sessions, "\n", fn session -> + "#{String.slice(session.id, 0, 8)} — #{session.title || "(untitled)"} (#{session.origin})" + end) + + {"Recent sessions:\n" <> body, state} + end + end + + defp run("/attach", "", _message, state), do: {"Say /attach .", state} + + defp run("/attach", argument, message, state) do + case resolve_session(argument) do + nil -> + {"No session starts with #{argument}.", state} + + session -> + state = Router.bind_state(state, message.adapter, message.conversation, session.id) + + {"Attached to #{String.slice(session.id, 0, 8)}. Both surfaces see this session now.", + state} + end + end + + defp run(decision, argument, message, state) when decision in ["/approve", "/deny"] do + {decide(decision, argument, message), state} + end + + defp run(unknown, _argument, _message, state), + do: {"I do not know #{unknown}. Say /help for the list.", state} + + # An approval answered from a channel: the request is found by the short form the person was + # shown, the cap decides whether a channel of this kind may answer a request of that tier at + # all, and only then does `Permissions.decide_request/3` make the decision. The cap is applied + # after the gate's own decision and never instead of it (docs/07); a capped answer is refused + # in words and receipted, because a silent refusal teaches a person the bot is broken. + defp decide(_command, "", _message), do: "Say /approve or /deny ." + + defp decide(command, argument, message) do + case find_pending(argument) do + nil -> + "No pending approval starts with #{argument}." + + approval -> + answer(command, approval, message) + end + end + + defp answer(command, approval, message) do + if Cap.allows?(message.adapter, approval.risk) do + decided(command, approval, message) + else + Receipts.gateway_cap_refusal(approval, message) + Cap.refusal(message.adapter, approval.risk) + end + end + + defp decided("/approve", approval, message) do + case Permissions.decide_request(approval.id, :once, by: decider(message)) do + {:ok, _} -> "Approved #{Format.short(approval.id)}." + {:error, reason} -> "Not decided: #{inspect(reason)}." + end + end + + defp decided("/deny", approval, message) do + case Permissions.decide_request(approval.id, :deny, by: decider(message)) do + {:ok, _} -> "Denied #{Format.short(approval.id)}." + {:error, reason} -> "Not decided: #{inspect(reason)}." + end + end + + # Who decided, as the approval row records it: the channel and the account, never "liveview". + defp decider(message), + do: "gateway:" <> Adapter.name(message.adapter) <> ":" <> message.external_user_id + + defp find_pending(argument) do + Enum.find(Permissions.pending(:all), &String.starts_with?(&1.id, argument)) + end + + # `/new` forgets the binding; the next message creates a session as a first message does. + defp forget(state, message) do + conv_key = {Adapter.name(message.adapter), message.conversation} + + case get_in(state.conversations, [conv_key, :session_id]) do + nil -> + state + + session_id -> + state + |> update_in([:conversations], &Map.delete(&1, conv_key)) + |> update_in([:sessions], &Map.delete(&1, session_id)) + end + end + + # A person types the short form they were shown, so a prefix resolves when it is unambiguous. + # The full id is only looked up when it is one: `get_session/1` casts, and a prefix reaches the + # repo as a malformed UUID and raises, which is a message from a channel crashing the router. + defp resolve_session(argument) do + if uuid?(argument), + do: Sessions.get_session(argument) || by_prefix(argument), + else: by_prefix(argument) + end + + defp uuid?(argument), + do: + Regex.match?(~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i, argument) + + defp by_prefix(prefix) do + case Enum.filter(Sessions.list_sessions(limit: 50), &String.starts_with?(&1.id, prefix)) do + [session] -> session + _ -> nil + end + end +end diff --git a/lib/trinity/gateways/console.ex b/lib/trinity/gateways/console.ex new file mode 100644 index 0000000..60100c8 --- /dev/null +++ b/lib/trinity/gateways/console.ex @@ -0,0 +1,110 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Console do + @moduledoc """ + The in-process gateway (slice 070): an adapter with no platform behind it. What it delivers is + kept in this process and read back by a test or by `mix trinity.console`, so every acceptance + criterion of this slice runs without an account, a token or a network, and so a developer can + talk to Trinity from a terminal on the same node. + + It is a real adapter and not a mock: the router treats it as it treats any other, and its + capabilities are the ones a plain text channel has (markdown, edits, a four thousand character + limit, no images, no buttons). A platform adapter (071, 072) differs from it in `deliver/2` and + `capabilities/0` and in nothing else. + """ + @behaviour Trinity.Gateways.Adapter + + use GenServer + + alias Trinity.Gateways.Format + + @capabilities %{markdown: true, images: false, buttons: false, edits: true, max_length: 4_000} + + ## The behaviour + + @impl Trinity.Gateways.Adapter + def capabilities, do: @capabilities + + @impl Trinity.Gateways.Adapter + def deliver(conversation, outbound), + do: GenServer.call(__MODULE__, {:deliver, conversation, outbound}) + + @impl Trinity.Gateways.Adapter + defdelegate format(text, capabilities), to: Format + + @impl Trinity.Gateways.Adapter + defdelegate render_approval(approval, capabilities), to: Format + + ## The process + + @doc "Starts the console's mailbox." + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts \\ []), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__) + + @impl GenServer + def init(_opts), do: {:ok, %{conversations: %{}, next_ref: 1}} + + @doc "Everything delivered to a conversation, oldest first." + @spec delivered(Trinity.Gateways.Adapter.conversation()) :: [ + Trinity.Gateways.Adapter.outbound() + ] + def delivered(conversation), do: GenServer.call(__MODULE__, {:delivered, conversation}) + + @doc """ + The text a conversation has been shown: every message and, for an edited message, the text it + last held. This is what a person in the channel would be looking at. + """ + @spec text(Trinity.Gateways.Adapter.conversation()) :: [String.t()] + def text(conversation), do: GenServer.call(__MODULE__, {:text, conversation}) + + @doc "Forgets a conversation's messages (a test's reset)." + @spec clear(Trinity.Gateways.Adapter.conversation()) :: :ok + def clear(conversation), do: GenServer.call(__MODULE__, {:clear, conversation}) + + @impl GenServer + def handle_call({:deliver, conversation, {:message, _} = outbound}, _from, state) do + # The reference is recorded beside the message, not recomputed when the log is read: the + # counter is global to this process and an edit names the reference `deliver/2` returned, so + # renumbering per conversation makes an edit land on a different message (found by the + # approvals suite, where a refusal overwrote an earlier reply). + ref = state.next_ref + state = append(state, conversation, {outbound, ref}) + {:reply, {:ok, ref}, %{state | next_ref: ref + 1}} + end + + def handle_call({:deliver, conversation, outbound}, _from, state), + do: {:reply, :ok, append(state, conversation, {outbound, nil})} + + def handle_call({:delivered, conversation}, _from, state), + do: + {:reply, state.conversations |> Map.get(conversation, []) |> Enum.map(&elem(&1, 0)), state} + + def handle_call({:text, conversation}, _from, state), + do: {:reply, state.conversations |> Map.get(conversation, []) |> visible(), state} + + def handle_call({:clear, conversation}, _from, state), + do: {:reply, :ok, %{state | conversations: Map.delete(state.conversations, conversation)}} + + defp append(state, conversation, entry) do + log = Map.get(state.conversations, conversation, []) + put_in(state.conversations[conversation], log ++ [entry]) + end + + # A message is what was sent; an edit replaces the message whose reference it names, which is + # how a stream looks to a person watching the channel rather than reading the log. An edit + # naming a message this conversation never had is ignored rather than invented. + defp visible(log) do + log + |> Enum.reduce({[], %{}}, fn + {{:message, text}, ref}, {order, by_ref} -> + {order ++ [ref], Map.put(by_ref, ref, text)} + + {{:edit, ref, text}, _}, {order, by_ref} -> + {order, if(Map.has_key?(by_ref, ref), do: Map.put(by_ref, ref, text), else: by_ref)} + + {{:typing, _}, _}, acc -> + acc + end) + |> then(fn {order, by_ref} -> Enum.map(order, &Map.fetch!(by_ref, &1)) end) + end +end diff --git a/lib/trinity/gateways/format.ex b/lib/trinity/gateways/format.ex new file mode 100644 index 0000000..50b372b --- /dev/null +++ b/lib/trinity/gateways/format.ex @@ -0,0 +1,132 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Format do + @moduledoc """ + Turning what Trinity says into what a channel can carry (slice 070): the default + `format/2` and `render_approval/2` every adapter delegates to unless its platform needs its own + dialect. + + Two jobs. **Chunking**: a platform refuses a message past its limit, so a long answer is split + at the latest paragraph, then line, then word boundary that fits, and only cut mid-word when a + single word is longer than the limit. A fenced code block that has to be split is re-opened on + the next chunk with its own fence, because half a fence renders the rest of the message as code. + **Plain text**: a channel without markdown gets the emphasis and heading markers removed rather + than shown; fences become plain lines. That is a reduction and it is named as one: it is not a + markdown renderer, and nothing here tries to be. + """ + + alias Trinity.Gateways.Adapter + alias Trinity.Permissions.Approval + + @fence "```" + + @doc "An assistant message as the channel's text, split to its length limit." + @spec format(String.t(), Adapter.capabilities()) :: [String.t()] + def format(text, capabilities) when is_binary(text) do + text + |> then(fn t -> if capabilities.markdown, do: t, else: plain(t) end) + |> chunk(capabilities.max_length) + end + + @doc """ + Splits text into pieces of at most `max` graphemes, at the latest boundary that fits. An open + code fence is closed at the end of a chunk and re-opened at the start of the next. + """ + @spec chunk(String.t(), pos_integer()) :: [String.t()] + def chunk(text, max) when is_binary(text) and is_integer(max) and max > 0 do + text + |> do_chunk(max, []) + |> Enum.reverse() + |> reopen_fences() + end + + defp do_chunk(text, max, acc) do + if String.length(text) <= max do + [text | acc] + else + {head, rest} = split_at_boundary(text, max) + do_chunk(String.trim_leading(rest), max, [head | acc]) + end + end + + # The latest paragraph, line or word boundary inside the limit; a hard cut when a single word + # is longer than the limit (a URL, a base64 blob) and there is no boundary to prefer. + defp split_at_boundary(text, max) do + head = String.slice(text, 0, max) + + case boundary(head) do + nil -> {head, String.slice(text, max..-1//1)} + at -> {String.trim_trailing(String.slice(head, 0, at)), String.slice(text, at..-1//1)} + end + end + + defp boundary(head) do + Enum.find_value([~r/\n\n(?!.*\n\n)/s, ~r/\n(?!.*\n)/s, ~r/ (?!.* )/s], fn re -> + case Regex.run(re, head, return: :index) do + [{start, len}] -> start + len + _ -> nil + end + end) + end + + # A chunk holding an odd number of fences left one open: close it, and open the next chunk with + # a fence of its own so the block keeps rendering as code. + defp reopen_fences(chunks) do + {out, _open} = + Enum.map_reduce(chunks, false, fn chunk, open? -> + chunk = if open?, do: @fence <> "\n" <> chunk, else: chunk + now_open? = chunk |> String.split(@fence) |> length() |> rem(2) == 0 + {if(now_open?, do: chunk <> "\n" <> @fence, else: chunk), now_open?} + end) + + out + end + + @doc "Markdown reduced to text for a channel that renders none of it." + @spec plain(String.t()) :: String.t() + def plain(text) when is_binary(text) do + text + |> String.replace(~r/^#{@fence}.*$/m, "") + |> String.replace(~r/^#+\s*/m, "") + |> String.replace(~r/\*\*([^*]+)\*\*/, "\\1") + |> String.replace(~r/(? String.replace(~r/(? String.replace(~r/^[ \t]*[-*][ \t]+/m, "• ") + |> String.replace(~r/\n{3,}/, "\n\n") + |> String.trim() + end + + @doc """ + An approval request as a message the conversation can answer. The commands are written out + whether or not the channel has buttons: a channel without them still has to be able to answer, + and `Trinity.Gateways.Commands` accepts the text form everywhere. + """ + @spec render_approval(Approval.t(), Adapter.capabilities()) :: Adapter.outbound() + def render_approval(%Approval{} = approval, capabilities) do + body = """ + Approval needed: #{approval.tool} (#{approval.risk}) + #{summarise(approval.args)} + Answer with /approve #{short(approval.id)} or /deny #{short(approval.id)} + """ + + {:message, body |> String.trim() |> chunk(capabilities.max_length) |> List.first()} + end + + @doc "The short form of an approval id a person can type back." + @spec short(String.t()) :: String.t() + def short(id) when is_binary(id), do: String.slice(id, 0, 8) + + defp summarise(args) when map_size(args) == 0, do: "(no arguments)" + + defp summarise(args) do + args + |> Enum.sort_by(fn {k, _} -> k end) + |> Enum.map_join(", ", fn {k, v} -> "#{k}=#{truncate(v)}" end) + end + + defp truncate(value) do + value |> to_string() |> String.slice(0, 40) + end +end diff --git a/lib/trinity/gateways/identities.ex b/lib/trinity/gateways/identities.ex new file mode 100644 index 0000000..bf340ae --- /dev/null +++ b/lib/trinity/gateways/identities.ex @@ -0,0 +1,188 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Identities do + @moduledoc """ + Who may talk to Trinity from outside (slice 070): the `gateway_identities` rows and the pairing + that creates them. + + The rule this module exists to hold, from docs/07: **an unknown sender gets a pairing prompt and + nothing else.** No session is created, no model is called, nothing is remembered but the code + that was shown. Pairing is proved on the desktop, where the owner already is: the code appears + on `/gateways`, the sender types it back into the channel, and only then does the row become + `paired`. A configured allowlist (`config :trinity, :gateways, allowlist: [{adapter, id}, …]`) + pairs an id at first sight, for a deployment that already knows who it is talking to. + + A code is six characters from an alphabet without the letters and digits that are read for one + another, lives ten minutes, and is compared in constant time. + """ + + import Ecto.Query, only: [from: 2] + + alias Trinity.Gateways.Identity + alias Trinity.Repo + + @alphabet ~c"ABCDEFGHJKLMNPQRSTUVWXYZ23456789" + @code_length 6 + @code_ttl_s 600 + + @doc "Every identity, newest first." + @spec list() :: [Identity.t()] + def list, do: Repo.all(from i in Identity, order_by: [desc: i.inserted_at]) + + @doc "The identity for an adapter and an external id, or nil." + @spec get(String.t(), String.t()) :: Identity.t() | nil + def get(adapter, external_user_id), + do: Repo.get_by(Identity, adapter: adapter, external_user_id: external_user_id) + + @doc """ + The identity an inbound message belongs to, creating a `pending` row with a fresh code when the + sender is unknown, or a `paired` one when the allowlist already names them. The second element + says what the caller must do: `:paired` to carry on, `:pending` to show the code and stop. + """ + @spec admit(String.t(), String.t(), keyword()) :: + {:ok, Identity.t(), :paired | :pending | :revoked} | {:error, Ecto.Changeset.t()} + def admit(adapter, external_user_id, opts \\ []) do + case get(adapter, external_user_id) do + nil -> first_sight(adapter, external_user_id, opts) + %Identity{state: "paired"} = identity -> {:ok, touch(identity, opts), :paired} + %Identity{state: "revoked"} = identity -> {:ok, identity, :revoked} + %Identity{} = identity -> {:ok, refresh_code(identity, opts), :pending} + end + end + + @doc """ + Answers a pairing attempt: the code the sender typed, against the code this identity was shown. + Wrong or expired leaves the row pending, and says so rather than saying which. + """ + @spec pair(String.t(), String.t(), String.t()) :: + {:ok, Identity.t()} | {:error, :no_code | :wrong_code | :expired | :revoked} + def pair(adapter, external_user_id, given) when is_binary(given) do + with %Identity{} = identity <- get(adapter, external_user_id), + :ok <- pairable(identity), + :ok <- code_matches(identity, given), + :ok <- code_live(identity) do + {:ok, + update!(identity, %{state: "paired", paired_at: now(), code: nil, code_expires_at: nil})} + else + nil -> {:error, :no_code} + {:error, _} = error -> error + end + end + + @doc "Turns an identity away; the row stays, so the refusal is visible." + @spec revoke(Identity.t()) :: {:ok, Identity.t()} | {:error, Ecto.Changeset.t()} + def revoke(%Identity{} = identity), + do: identity |> Identity.changeset(%{state: "revoked", revoked_at: now()}) |> Repo.update() + + @doc "Pairs an identity from the desktop, without a code (the owner is already proved here)." + @spec allow(Identity.t()) :: {:ok, Identity.t()} | {:error, Ecto.Changeset.t()} + def allow(%Identity{} = identity) do + identity + |> Identity.changeset(%{state: "paired", paired_at: now(), code: nil, code_expires_at: nil}) + |> Repo.update() + end + + @doc "The identities waiting with a live code, for the page that shows them." + @spec pending() :: [Identity.t()] + def pending do + now = now() + + Repo.all( + from i in Identity, + where: i.state == "pending" and i.code_expires_at > ^now, + order_by: [desc: i.inserted_at] + ) + end + + @doc "A fresh pairing code." + @spec generate_code() :: String.t() + def generate_code do + for _ <- 1..@code_length, into: "", do: <> + end + + @doc "How long a code lives, in seconds." + @spec code_ttl_s() :: pos_integer() + def code_ttl_s, do: @code_ttl_s + + defp first_sight(adapter, external_user_id, opts) do + attrs = %{ + adapter: adapter, + external_user_id: external_user_id, + display_name: Keyword.get(opts, :display_name), + last_conversation: Keyword.get(opts, :conversation) + } + + if allowlisted?(adapter, external_user_id) do + insert(Map.merge(attrs, %{state: "paired", paired_at: now()}), :paired) + else + insert( + Map.merge(attrs, %{state: "pending", code: generate_code(), code_expires_at: expiry()}), + :pending + ) + end + end + + defp insert(attrs, state) do + case %Identity{} |> Identity.changeset(attrs) |> Repo.insert() do + {:ok, identity} -> {:ok, identity, state} + {:error, _} = error -> error + end + end + + # A pending identity that comes back after its code expired is shown a new one: the alternative + # is a sender who can never pair because the first code they were given has gone stale. + defp refresh_code(identity, opts) do + if live_code?(identity) do + touch(identity, opts) + else + update!(identity, %{ + code: generate_code(), + code_expires_at: expiry(), + last_conversation: Keyword.get(opts, :conversation, identity.last_conversation) + }) + end + end + + defp touch(identity, opts) do + case Keyword.get(opts, :conversation) do + nil -> identity + same when same == identity.last_conversation -> identity + conversation -> update!(identity, %{last_conversation: conversation}) + end + end + + defp update!(identity, attrs), do: identity |> Identity.changeset(attrs) |> Repo.update!() + + defp pairable(%Identity{state: "revoked"}), do: {:error, :revoked} + defp pairable(%Identity{state: "paired"} = _identity), do: :ok + defp pairable(%Identity{code: nil}), do: {:error, :no_code} + defp pairable(%Identity{}), do: :ok + + defp code_matches(%Identity{code: nil}, _given), do: {:error, :no_code} + + defp code_matches(%Identity{code: code}, given) do + given = given |> String.trim() |> String.upcase() + + if byte_size(given) == byte_size(code) and :crypto.hash_equals(given, code), + do: :ok, + else: {:error, :wrong_code} + end + + defp code_live(identity), do: if(live_code?(identity), do: :ok, else: {:error, :expired}) + + defp live_code?(%Identity{code: code, code_expires_at: at}), + do: is_binary(code) and not is_nil(at) and DateTime.compare(at, now()) == :gt + + defp allowlisted?(adapter, external_user_id) do + :trinity + |> Application.get_env(:gateways, []) + |> Keyword.get(:allowlist, []) + |> Enum.any?(fn + {a, id} -> to_string(a) == adapter and to_string(id) == external_user_id + _ -> false + end) + end + + defp expiry, do: DateTime.add(now(), @code_ttl_s, :second) + defp now, do: DateTime.utc_now() +end diff --git a/lib/trinity/gateways/identity.ex b/lib/trinity/gateways/identity.ex new file mode 100644 index 0000000..86ef7c5 --- /dev/null +++ b/lib/trinity/gateways/identity.ex @@ -0,0 +1,67 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Identity do + @moduledoc """ + One row of `gateway_identities` (slice 070, docs/05, docs/07 "Gateways"): a person on one + outside platform and what Trinity will do for them. The key is the pair `(adapter, + external_user_id)`, because an id means nothing outside the platform that issued it. + + `state` is `pending` (a pairing code has been shown and not yet entered), `paired` (the owner + typed the code into the desktop's channel, which is the proof) or `revoked`. A revoked row is + kept rather than deleted: an id that was turned away should be visible on the page, not absent + from it, and pairing again is then a deliberate act with a record. + """ + use Ecto.Schema + + import Ecto.Changeset + + @primary_key {:id, Trinity.UUID, autogenerate: true} + @foreign_key_type Trinity.UUID + @timestamps_opts [type: :utc_datetime_usec] + + @type t :: %__MODULE__{} + + @states ~w(pending paired revoked) + + schema "gateway_identities" do + field :adapter, :string + field :external_user_id, :string + field :display_name, :string + field :state, :string, default: "pending" + field :code, :string + field :code_expires_at, :utc_datetime_usec + field :paired_at, :utc_datetime_usec + field :revoked_at, :utc_datetime_usec + field :last_conversation, :string + timestamps() + end + + @doc "The states a row may be in." + @spec states() :: [String.t()] + def states, do: @states + + @doc "A changeset for an identity." + @spec changeset(t(), map()) :: Ecto.Changeset.t() + def changeset(identity, attrs) do + identity + |> cast(attrs, [ + :adapter, + :external_user_id, + :display_name, + :state, + :code, + :code_expires_at, + :paired_at, + :revoked_at, + :last_conversation + ]) + |> validate_required([:adapter, :external_user_id, :state]) + |> validate_inclusion(:state, @states) + |> unique_constraint([:adapter, :external_user_id]) + end + + @doc "True when this identity may talk to Trinity right now." + @spec paired?(t()) :: boolean() + def paired?(%__MODULE__{state: "paired"}), do: true + def paired?(%__MODULE__{}), do: false +end diff --git a/lib/trinity/gateways/receipts.ex b/lib/trinity/gateways/receipts.ex new file mode 100644 index 0000000..6c55ead --- /dev/null +++ b/lib/trinity/gateways/receipts.ex @@ -0,0 +1,49 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Receipts do + @moduledoc """ + What a gateway writes to the receipt chain (slice 070). One thing, for now: a decision that a + channel was not allowed to make. docs/07 says a capped request is refused rather than dropped, + and a refusal nobody can find afterwards is a drop with a message attached, so it is receipted + on the session's own chain beside the decisions the gate writes there. + + The channel and the account that tried are named; nothing the person typed is copied, because + the text of a message to a bot is not evidence about an approval. + """ + + alias Trinity.Gateways.{Adapter, Cap} + alias Trinity.Permissions.Approval + alias Trinity.Receipts + + require Logger + + @doc "Receipts a decision the channel's cap refused to make." + @spec gateway_cap_refusal(Approval.t(), map()) :: :ok + def gateway_cap_refusal(%Approval{} = approval, message) do + adapter = Adapter.name(message.adapter) + + Receipts.append(Receipts.session_scope(approval.session_id), %{ + kind: "decision", + subject: %{ + "phase" => "gateway_cap", + "adapter" => adapter, + "external_user_id" => message.external_user_id, + "conversation" => message.conversation, + "tool" => approval.tool, + "risk" => approval.risk, + "origin" => "gateway" + }, + decision: %{ + "outcome" => "deny", + "basis" => "channel_cap", + "reason" => "#{approval.risk} is above this channel's ceiling #{Cap.ceiling(adapter)}" + }, + subject_ref: "approval:" <> approval.id, + meta: %{} + }) + + :ok + rescue + error -> Logger.warning("gateway: cap refusal not receipted: #{Exception.message(error)}") + end +end diff --git a/lib/trinity/gateways/router.ex b/lib/trinity/gateways/router.ex new file mode 100644 index 0000000..a801b57 --- /dev/null +++ b/lib/trinity/gateways/router.ex @@ -0,0 +1,427 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.Router do + @moduledoc """ + The one road between a channel and Trinity (slice 070). An adapter hands it an inbound message + and it decides, in this order and no other: is this identity admitted (`Identities`), is it + inside its rate limit, is the text a command (`Commands`), and otherwise whose session does it + belong to. Then it sends the text to that session and streams the answer back to the channel. + + Why the order matters. An unknown sender reaches the pairing check and stops there: no session + is created and no model is called for someone who has not been let in, which is docs/07's rule + rather than an implementation detail. The rate limit sits above the session so a flood costs a + row lookup and nothing else. + + **Streaming.** The router subscribes to `session:` once per conversation and coalesces + `{:assistant_delta, _}` into one message it edits while the turn runs, when the adapter can + edit; a channel that cannot edit is sent the finished message once, because a message per delta + is unreadable and, on a real platform, rate-limited into oblivion. The turn's final + `{:assistant_message, _}` is always delivered, so a channel sees the whole answer whatever its + capabilities. + """ + use GenServer + + alias Trinity.Gateways.{Adapter, Cap, Commands, Format, Identities} + alias Trinity.Permissions + alias Trinity.Sessions + + require Logger + + @flush_ms 700 + @default_rate [capacity: 20, per_minute: 20] + + @typedoc "What an adapter hands in: who said what, where." + @type inbound :: %{ + adapter: module(), + conversation: Adapter.conversation(), + external_user_id: String.t(), + text: String.t(), + display_name: String.t() | nil + } + + ## The API an adapter uses + + @doc "Starts the router." + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts \\ []), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__) + + @doc """ + One inbound message. Answers once the message has been placed (the reply arrives on the channel + as the turn streams), or with why it was refused: `{:error, :pending}` when a pairing code was + shown instead, `{:error, :revoked}`, `{:error, :rate_limited}`. + """ + @spec inbound(module(), Adapter.conversation(), String.t(), String.t(), keyword()) :: + {:ok, :placed | :command | :paired} | {:error, term()} + def inbound(adapter, conversation, external_user_id, text, opts \\ []) do + message = %{ + adapter: adapter, + conversation: conversation, + external_user_id: external_user_id, + text: text, + display_name: Keyword.get(opts, :display_name) + } + + GenServer.call(__MODULE__, {:inbound, message}, 30_000) + end + + @doc "The session a conversation is bound to, or nil." + @spec session_of(module(), Adapter.conversation()) :: String.t() | nil + def session_of(adapter, conversation), + do: GenServer.call(__MODULE__, {:session_of, adapter, conversation}) + + @doc """ + Binds a conversation to a session that already exists (`/attach`), so the desktop and the + channel watch one stream. The conversation's own session, if it had one, is left alone. + """ + @spec attach(module(), Adapter.conversation(), String.t()) :: :ok | {:error, :no_session} + def attach(adapter, conversation, session_id), + do: GenServer.call(__MODULE__, {:attach, adapter, conversation, session_id}) + + @doc "Sends an outbound instruction to a conversation through its adapter (used by delivery)." + @spec deliver(module(), Adapter.conversation(), Adapter.outbound()) :: :ok + def deliver(adapter, conversation, outbound) do + _ = adapter.deliver(conversation, outbound) + :ok + end + + ## The process + + @impl GenServer + def init(_opts) do + {:ok, %{conversations: %{}, sessions: %{}, buckets: %{}}} + end + + @impl GenServer + def handle_call({:inbound, message}, _from, state) do + case admit(message) do + {:ok, :paired} -> route(message, state) + {:refused, reply, result} -> {:reply, result, say(message, reply, state)} + end + end + + def handle_call({:session_of, adapter, conversation}, _from, state), + do: {:reply, get_in(state.conversations, [key(adapter, conversation), :session_id]), state} + + def handle_call({:attach, adapter, conversation, session_id}, _from, state) do + case Sessions.get_session(session_id) do + nil -> {:reply, {:error, :no_session}, state} + _row -> {:reply, :ok, bind_state(state, adapter, conversation, session_id)} + end + end + + # The stream: everything on `session:` for a bound conversation. + @impl GenServer + def handle_info({:session, session_id, event}, state) do + case Map.get(state.sessions, session_id) do + nil -> {:noreply, state} + conv_key -> {:noreply, stream(event, conv_key, state)} + end + end + + def handle_info({:approval, :requested, approval}, state) do + {:noreply, render_approval(approval, state)} + end + + def handle_info({:approval, :decided, _approval}, state), do: {:noreply, state} + + def handle_info({:flush, conv_key}, state), do: {:noreply, flush(conv_key, state)} + def handle_info(_other, state), do: {:noreply, state} + + ## Admission + + defp admit(message) do + adapter_name = Adapter.name(message.adapter) + + case Identities.admit(adapter_name, message.external_user_id, + conversation: message.conversation, + display_name: message.display_name + ) do + {:ok, _identity, :paired} -> + {:ok, :paired} + + {:ok, identity, :pending} -> + pairing_reply(adapter_name, message, identity) + + {:ok, _identity, :revoked} -> + {:refused, "This identity is not allowed to use Trinity.", {:error, :revoked}} + + {:error, _changeset} -> + {:refused, "Something went wrong admitting you.", {:error, :not_admitted}} + end + end + + # A pending sender's message is never a message: it is either the code, or it is answered with + # the prompt again. Nothing else of it is read. + defp pairing_reply(adapter_name, message, identity) do + case Identities.pair(adapter_name, message.external_user_id, message.text) do + {:ok, _paired} -> + {:refused, "Paired. Say anything and Trinity will answer.", {:ok, :paired}} + + {:error, _reason} -> + {:refused, prompt_for(identity), {:error, :pending}} + end + end + + defp prompt_for(identity) do + """ + Trinity does not know you yet. Open the desktop app, go to Settings → Gateways, and you will + see the code #{identity.code}. Send that code here to pair. It lasts #{div(Identities.code_ttl_s(), 60)} minutes. + """ + |> String.trim() + end + + ## Routing an admitted message + + defp route(message, state) do + case take_token(state, message) do + {:error, state} -> + {:reply, {:error, :rate_limited}, + say(message, "Too many messages. Wait a moment.", state)} + + {:ok, state} -> + dispatch(message, state) + end + end + + # A command runs for text an outside channel chose, and this process holds every conversation's + # binding: a command that raises must cost its own message and nothing else. The reason is + # logged and the channel is told, rather than the router dying and every conversation with it. + defp dispatch(%{text: "/" <> _} = message, state) do + {reply, state} = Commands.handle(message, state) + {:reply, {:ok, :command}, say(message, reply, state)} + rescue + error -> + Logger.error("gateway command failed: #{Exception.message(error)}") + {:reply, {:error, :command_failed}, say(message, "That command did not work.", state)} + end + + defp dispatch(message, state) do + {session_id, state} = session_for(message, state) + + case Sessions.send_user_message(session_id, message.text) do + {:ok, _} -> + {:reply, {:ok, :placed}, state} + + {:error, reason} -> + Logger.warning("gateway: message refused: #{inspect(reason)}") + + {:reply, {:error, reason}, + say(message, "Trinity is busy with the previous message.", state)} + end + end + + ## Sessions and binding + + defp session_for(message, state) do + conv_key = key(message.adapter, message.conversation) + + case get_in(state.conversations, [conv_key, :session_id]) do + nil -> + {:ok, row} = create_session(message) + {row.id, bind_state(state, message.adapter, message.conversation, row.id)} + + session_id -> + {session_id, state} + end + end + + defp create_session(message) do + Sessions.create_session(%{ + persona_id: Sessions.default_persona().id, + origin: Adapter.name(message.adapter), + title: "#{Adapter.name(message.adapter)}:#{message.conversation}", + origin_ref: %{ + "adapter" => Adapter.name(message.adapter), + "conversation" => message.conversation, + "external_user_id" => message.external_user_id + } + }) + end + + @doc """ + Binds a conversation to a session, as a state transition. Public because `Commands` runs inside + this process: a command that called `attach/3` would be this process calling itself, which is a + deadlock the suite found. Callers outside the process use `attach/3`. + """ + @spec bind_state(map(), module(), Adapter.conversation(), String.t()) :: map() + def bind_state(state, adapter, conversation, session_id) do + conv_key = key(adapter, conversation) + :ok = Sessions.subscribe(session_id) + # The approvals a turn in this session raises are rendered into the conversation that + # started it, so the person who asked is the person who answers (docs/07's approval channel). + :ok = Permissions.subscribe(session_id) + + conversation_state = %{ + adapter: adapter, + conversation: conversation, + session_id: session_id, + buffer: "", + ref: nil, + timer: nil + } + + state + |> put_in([:conversations, conv_key], conversation_state) + |> put_in([:sessions, session_id], conv_key) + end + + ## The stream to the channel + + defp stream({:assistant_delta, delta}, conv_key, state) do + conv = Map.fetch!(state.conversations, conv_key) + + if conv.adapter.capabilities().edits do + state + |> put_in([:conversations, conv_key, :buffer], conv.buffer <> delta) + |> schedule_flush(conv_key) + else + state + end + end + + defp stream({:assistant_message, message}, conv_key, state) do + conv = Map.fetch!(state.conversations, conv_key) + state = cancel_timer(state, conv_key) + text = message.content || conv.buffer + + for chunk <- + conv.adapter.format(text, conv.adapter.capabilities()) |> then(&edit_or_send(conv, &1)) do + chunk + end + + state + |> put_in([:conversations, conv_key, :buffer], "") + |> put_in([:conversations, conv_key, :ref], nil) + end + + defp stream({:error, reason}, conv_key, state) do + conv = Map.fetch!(state.conversations, conv_key) + + _ = + conv.adapter.deliver( + conv.conversation, + {:message, "Trinity hit an error: #{inspect(reason)}"} + ) + + state + end + + defp stream(_event, _conv_key, state), do: state + + # The finished answer: the message being edited becomes the first chunk, the rest are new + # messages, so a long answer arrives whole on a channel with a length limit. + defp edit_or_send(conv, [first | rest]) do + case conv.ref do + nil -> _ = conv.adapter.deliver(conv.conversation, {:message, first}) + ref -> _ = conv.adapter.deliver(conv.conversation, {:edit, ref, first}) + end + + for chunk <- rest, do: conv.adapter.deliver(conv.conversation, {:message, chunk}) + end + + defp edit_or_send(_conv, []), do: [] + + defp schedule_flush(state, conv_key) do + case get_in(state.conversations, [conv_key, :timer]) do + nil -> + timer = Process.send_after(self(), {:flush, conv_key}, @flush_ms) + put_in(state.conversations[conv_key][:timer], timer) + + _running -> + state + end + end + + defp flush(conv_key, state) do + case Map.get(state.conversations, conv_key) do + nil -> state + %{buffer: ""} -> put_in(state.conversations[conv_key][:timer], nil) + conv -> flush_buffer(conv, conv_key, state) + end + end + + defp flush_buffer(conv, conv_key, state) do + [partial | _] = conv.adapter.format(conv.buffer, conv.adapter.capabilities()) + + ref = + case conv.ref do + nil -> + case conv.adapter.deliver(conv.conversation, {:message, partial}) do + {:ok, ref} -> ref + _ -> nil + end + + ref -> + _ = conv.adapter.deliver(conv.conversation, {:edit, ref, partial}) + ref + end + + state + |> put_in([:conversations, conv_key, :ref], ref) + |> put_in([:conversations, conv_key, :timer], nil) + end + + defp cancel_timer(state, conv_key) do + case get_in(state.conversations, [conv_key, :timer]) do + nil -> state + timer -> Process.cancel_timer(timer) && put_in(state.conversations[conv_key][:timer], nil) + end + end + + # A request is rendered into the conversation bound to its session, and nowhere else: an + # approval belongs to the surface that asked for it. A tier this channel may not answer is said + # so here, so the person is not left typing /approve at something that will refuse them. + defp render_approval(approval, state) do + case Map.get(state.sessions, approval.session_id) do + nil -> + state + + conv_key -> + conv = Map.fetch!(state.conversations, conv_key) + {:message, text} = conv.adapter.render_approval(approval, conv.adapter.capabilities()) + + text = + if Cap.allows?(conv.adapter, approval.risk), + do: text, + else: text <> "\n\n" <> Cap.refusal(conv.adapter, approval.risk) + + _ = conv.adapter.deliver(conv.conversation, {:message, text}) + state + end + end + + ## Rate limiting: a token bucket per identity, in this process's own state + + defp take_token(state, message) do + config = Keyword.merge(@default_rate, rate_config()) + capacity = Keyword.fetch!(config, :capacity) + per_minute = Keyword.fetch!(config, :per_minute) + bucket_key = {Adapter.name(message.adapter), message.external_user_id} + now = System.monotonic_time(:millisecond) + {tokens, last} = Map.get(state.buckets, bucket_key, {capacity, now}) + refilled = min(capacity, tokens + (now - last) * per_minute / 60_000) + + if refilled >= 1 do + {:ok, put_in(state.buckets[bucket_key], {refilled - 1, now})} + else + {:error, put_in(state.buckets[bucket_key], {refilled, now})} + end + end + + defp rate_config, + do: :trinity |> Application.get_env(:gateways, []) |> Keyword.get(:rate_limit, []) + + ## Saying something back without a session + + defp say(message, text, state) do + caps = message.adapter.capabilities() + + for chunk <- Format.chunk(text, caps.max_length) do + message.adapter.deliver(message.conversation, {:message, chunk}) + end + + state + end + + defp key(adapter, conversation), do: {Adapter.name(adapter), conversation} +end diff --git a/lib/trinity/scheduler.ex b/lib/trinity/scheduler.ex index e282bee..9ebe877 100644 --- a/lib/trinity/scheduler.ex +++ b/lib/trinity/scheduler.ex @@ -15,8 +15,19 @@ defmodule Trinity.Scheduler do `tasks` carries `{:task_run, %Run{}}` when a run is delivered. """ use Boundary, - deps: [Trinity, Trinity.Sessions, Trinity.LLM], - exports: [Task, Run, Delivery, Delivery.Desktop, Parse, Workers.Tick, Workers.RunTask] + # Slice 070: a run can be delivered to a channel, which is the one thing the scheduler asks + # of the gateway layer (docs/01's row said so when 050 was built). + deps: [Trinity, Trinity.Sessions, Trinity.LLM, Trinity.Gateways], + exports: [ + Task, + Run, + Delivery, + Delivery.Desktop, + Delivery.Gateway, + Parse, + Workers.Tick, + Workers.RunTask + ] import Ecto.Query, only: [from: 2] diff --git a/lib/trinity/scheduler/delivery/gateway.ex b/lib/trinity/scheduler/delivery/gateway.ex new file mode 100644 index 0000000..9df85bc --- /dev/null +++ b/lib/trinity/scheduler/delivery/gateway.ex @@ -0,0 +1,66 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Scheduler.Delivery.Gateway do + @moduledoc """ + A finished run delivered to a channel (slice 070, the second implementation of slice 050's + `Trinity.Scheduler.Delivery`). A task names its destination in `deliver_to`: + + %{"kind" => "gateway", "adapter" => "console", "conversation" => "c-1"} + + The run's summary is sent through the adapter's own formatting, so a long one is chunked to the + channel's limit like any other message. The run is marked delivered and broadcast as the + desktop delivery does, so the tasks page still shows it: a result that went to a channel is not + a result the desktop should forget. + + A destination naming an adapter that is not configured is an error rather than a silent drop, + and the run keeps its undelivered state for the page to show. + """ + @behaviour Trinity.Scheduler.Delivery + + alias Trinity.Gateways + alias Trinity.Gateways.Adapter + alias Trinity.Scheduler + alias Trinity.Scheduler.{Run, Task} + + @impl true + def deliver(%Run{} = run, %Task{deliver_to: destination} = task) do + with {:ok, adapter} <- adapter(destination), + {:ok, conversation} <- fetch(destination, "conversation") do + send_summary(adapter, conversation, run, task) + mark(run) + end + end + + defp send_summary(adapter, conversation, run, task) do + text = "#{task.name}: #{run.summary || "(no summary)"}" + + for chunk <- adapter.format(text, adapter.capabilities()) do + adapter.deliver(conversation, {:message, chunk}) + end + end + + defp mark(run) do + with {:ok, run} <- Scheduler.update_run(run, %{delivered_at: DateTime.utc_now()}) do + Scheduler.broadcast(run) + {:ok, run} + end + end + + # The adapter is named, never handed in: a task row is data, and a module name in a row that + # the scheduler would call is a way to run any module by writing a row. + defp adapter(destination) do + with {:ok, name} <- fetch(destination, "adapter") do + case Enum.find(Gateways.adapters(), &(Adapter.name(&1) == name)) do + nil -> {:error, {:unknown_adapter, name}} + adapter -> {:ok, adapter} + end + end + end + + defp fetch(destination, key) do + case Map.get(destination, key) do + value when is_binary(value) and value != "" -> {:ok, value} + _ -> {:error, {:missing, key}} + end + end +end diff --git a/lib/trinity_web/live/gateways_live.ex b/lib/trinity_web/live/gateways_live.ex new file mode 100644 index 0000000..2148f8d --- /dev/null +++ b/lib/trinity_web/live/gateways_live.ex @@ -0,0 +1,204 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule TrinityWeb.GatewaysLive do + @moduledoc """ + `/gateways` (slice 070): the channels Trinity can be reached from, who is waiting to be paired, + and who has been let in. + + The pairing code is shown here and nowhere else, which is the whole of the pairing proof: a + sender who can read this page is at the owner's desktop. Each identity can be allowed without a + code or revoked, and a revoked row stays visible rather than disappearing, so turning someone + away is a thing the page records rather than a thing it forgets. The tier ceiling each adapter + carries is shown beside it, because what a channel may approve is the question this page is + really about (docs/07, the channel trust cap). + """ + use TrinityWeb, :live_view + + alias Trinity.Gateways + alias Trinity.Gateways.{Adapter, Cap, Identities} + + @refresh_ms 2_000 + + @impl true + def mount(_params, _session, socket) do + if connected?(socket), do: Process.send_after(self(), :refresh, @refresh_ms) + {:ok, socket |> assign(page_title: gettext("Gateways")) |> load()} + end + + defp load(socket) do + assign(socket, + adapters: adapters(), + identities: Identities.list(), + ttl_minutes: div(Identities.code_ttl_s(), 60) + ) + end + + defp adapters do + for adapter <- Gateways.adapters() do + %{ + module: adapter, + name: Adapter.name(adapter), + running?: is_pid(Process.whereis(adapter)), + ceiling: Cap.ceiling(adapter), + capabilities: adapter.capabilities() + } + end + end + + @impl true + def handle_info(:refresh, socket) do + Process.send_after(self(), :refresh, @refresh_ms) + {:noreply, load(socket)} + end + + @impl true + def handle_event("allow", %{"id" => id}, socket), + do: {:noreply, act(socket, id, &Identities.allow/1)} + + def handle_event("revoke", %{"id" => id}, socket), + do: {:noreply, act(socket, id, &Identities.revoke/1)} + + defp act(socket, id, fun) do + case Enum.find(socket.assigns.identities, &(&1.id == id)) do + nil -> put_flash(socket, :error, gettext("That identity is gone.")) + identity -> with({:ok, _} <- fun.(identity), do: load(socket)) + end + end + + @impl true + def render(assigns) do + ~H""" + + <:bar> + {gettext("Gateways")} + {length(@identities)} + +
+
+

{gettext("Channels")}

+

+ {gettext( + "No gateway is configured. A channel is a module in config :trinity, :gateways, adapters: [...]; the console adapter ships with Trinity and is what mix trinity.console talks to." + )} +

+
    +
  • + {adapter.name} + + {if adapter.running?, do: gettext("running"), else: gettext("not started")} + + + {gettext("approves up to")} + {adapter.ceiling} + + + {gettext("max")} + {adapter.capabilities.max_length} + +
    • +
+

+ {gettext( + "A channel may approve up to its ceiling and no further: an exec or destructive request is decided on the desktop, on the permissions page. The gate still decides every call whatever the channel." + )} +

+
+ +
+

{gettext("Waiting to pair")}

+

+ {gettext("Nobody is waiting. An unknown sender is shown a code here when they write.")} +

+
    +
  • + {identity.code} + + {identity.adapter}:{identity.external_user_id} + + + {identity.display_name} + + + + {gettext("expires in %{n} minutes", n: @ttl_minutes)} + + + +
    • +
+

+ {gettext( + "Send the code from the channel to pair, or allow it here. Until then that sender gets the pairing prompt and nothing else: no session, no model." + )} +

+
+ +
+

{gettext("Identities")}

+ + + + + + + + + + +
{identity.adapter}{identity.external_user_id}{identity.state}{identity.display_name} + + +
+

+ {gettext("No identity has written yet.")} +

+
+
+ + """ + end + + defp pending(identities), do: Enum.filter(identities, &(&1.state == "pending" and &1.code)) +end diff --git a/lib/trinity_web/router.ex b/lib/trinity_web/router.ex index 229c166..e55bc1f 100644 --- a/lib/trinity_web/router.ex +++ b/lib/trinity_web/router.ex @@ -57,6 +57,8 @@ defmodule TrinityWeb.Router do live "/skills", SkillsLive, :index # Slice 060: the MCP servers and their health. live "/mcp", MCPLive, :index + # Slice 070: the channels Trinity can be reached from, and who may. + live "/gateways", GatewaysLive, :index # Slice 050: scheduled tasks, their runs and the results to read. live "/tasks", TasksLive, :index # Slice 024: a session's receipt chain, and the boot receipt of this run. diff --git a/priv/repo/migrations/20260922200000_create_gateway_identities.exs b/priv/repo/migrations/20260922200000_create_gateway_identities.exs new file mode 100644 index 0000000..802b3a0 --- /dev/null +++ b/priv/repo/migrations/20260922200000_create_gateway_identities.exs @@ -0,0 +1,30 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Repo.Migrations.CreateGatewayIdentities do + use Ecto.Migration + + # Slice 070 (docs/05, docs/07 "Gateways"): who outside this machine may talk to Trinity. One + # row per (adapter, external user), unique on the pair: an id is only ever known within the + # platform that issued it, so the adapter is half the key. `state` is "pending" (a code was + # shown and not yet entered), "paired" (the owner proved it) or "revoked" (kept rather than + # deleted, so a revoked id cannot silently re-pair itself without appearing here). + def change do + create table(:gateway_identities, primary_key: false) do + add :id, :binary_id, primary_key: true + add :adapter, :string, null: false + add :external_user_id, :string, null: false + add :display_name, :string + add :state, :string, null: false, default: "pending" + add :code, :string + add :code_expires_at, :utc_datetime_usec + add :paired_at, :utc_datetime_usec + add :revoked_at, :utc_datetime_usec + # The conversation the pairing was asked from, so the answer goes back where it came from. + add :last_conversation, :string + timestamps(type: :utc_datetime_usec) + end + + create unique_index(:gateway_identities, [:adapter, :external_user_id]) + create index(:gateway_identities, [:state]) + end +end diff --git a/scripts/dev_gateways.sh b/scripts/dev_gateways.sh new file mode 100755 index 0000000..d6e81cd --- /dev/null +++ b/scripts/dev_gateways.sh @@ -0,0 +1,76 @@ +#!/usr/bin/env bash +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +# Slice 070's manual queue (AC2, AC9): the application in the test environment on its own +# database (trinity_screenshots.db, never the suite's), with the console gateway configured and +# the fake provider answering, so `/gateways` can be photographed and the pairing flow read as a +# transcript. It runs the flow through `Trinity.Gateways.Router` itself, which is the same path a +# platform adapter takes; nothing here is a mock of the gateway layer. +# +# Prints the transcript, then PORT= and stays up for the browser. Kill it by the pid +# `pgrep -af beam.smp` prints. Run `mix assets.build` first. +cd "$(dirname "$0")/.." +export MIX_ENV=test +exec systemd-run --user --scope -p MemoryMax=32G --quiet -- mix run --no-start --no-halt -e ' +repo = Application.get_env(:trinity, Trinity.Repo) |> Keyword.delete(:pool) |> Keyword.put(:database, "trinity_screenshots.db") +Application.put_env(:trinity, Trinity.Repo, repo) +rrepo = Application.get_env(:trinity, Trinity.Repo.Receipts) |> Keyword.delete(:pool) |> Keyword.put(:database, "trinity_screenshots_receipts.db") +Application.put_env(:trinity, Trinity.Repo.Receipts, rrepo) +endpoint = Application.get_env(:trinity, TrinityWeb.Endpoint) |> Keyword.put(:check_origin, false) +Application.put_env(:trinity, TrinityWeb.Endpoint, endpoint) +Application.put_env(:trinity, :mcp_boot, false) +Application.put_env(:trinity, :gateways, adapters: [Trinity.Gateways.Console]) +{:ok, _} = Application.ensure_all_started(:ecto_sql) +for r <- [Trinity.Repo, Trinity.Repo.Receipts] do + {:ok, pid} = r.start_link() + Ecto.Migrator.run(r, :up, all: true) + GenServer.stop(pid) +end +{:ok, _} = Application.ensure_all_started(:trinity) +# The transcript is the point of this script, so the query log does not share the terminal. +Logger.configure(level: :warning) + +alias Trinity.Gateways.{Console, Identities, Router} +{:ok, _} = Console.start_link([]) +{:ok, _} = Router.start_link([]) + +# The fake provider answers, so the transcript is a real turn without a key. +Trinity.LLM.Providers.Fake.scripts([ + Enum.flat_map(String.split("Yes: three tasks are due today, and the notes directory changed."), &[{:text_delta, &1 <> " "}, {:sleep, 20}]) ++ + [{:usage, %{input_tokens: 12, output_tokens: 20}}, {:done, :stop}] +]) + +conv = "phone" +user = "+15551234567" +say = fn text -> + before = length(Console.delivered(conv)) + result = Router.inbound(Console, conv, user, text, display_name: "Ayla (phone)") + Process.sleep(1200) + shown = Console.text(conv) |> Enum.drop(Enum.count(Enum.take(Console.delivered(conv), before), &match?({:message, _}, &1))) + IO.puts("\n they say: #{text}") + IO.puts(" result: #{inspect(result)}") + for line <- shown, do: IO.puts(" trinity: " <> String.replace(line, "\n", "\n ")) + :ok +end + +IO.puts("\n=== AC2: an unknown sender is answered with a pairing code and nothing else ===") +say.("are my tasks done?") +identity = Identities.get("console", user) +IO.puts("\n sessions created so far: #{length(Trinity.Sessions.list_sessions())} (a stranger gets no session)") +IO.puts(" the page shows the code: #{identity.code}") + +IO.puts("\n=== the wrong code is refused, and says only that ===") +say.("AAAAAA") + +IO.puts("\n=== the code from /gateways pairs them ===") +say.(identity.code) + +IO.puts("\n=== and the next message is a message ===") +say.("are my tasks done?") +IO.puts("\n sessions now: #{length(Trinity.Sessions.list_sessions())}") + +# A second identity is left pending, so the page has a code to photograph. +Router.inbound(Console, "laptop", "u-waiting", "hello?", display_name: "A new device") +{:ok, {_, port}} = TrinityWeb.Endpoint.server_info(:http) +IO.puts("\nPORT=#{port}") +' diff --git a/slices/070-gateway-core/NOTES.md b/slices/070-gateway-core/NOTES.md new file mode 100644 index 0000000..e0afae2 --- /dev/null +++ b/slices/070-gateway-core/NOTES.md @@ -0,0 +1,135 @@ +# Slice 070: NOTES + +## Read before code, 2026-09-22 + +Tree at `8b31ee5` on `main` (062 approved; **M5a Automates reached**; 012, the dependency, +approved since 2026-09-20). Chosen from the ready set by the coding agent on the owner's +instruction to pick, and why it rather than the others: 080 and 090 also have approved +dependencies, but 070 unblocks two slices (071, 072) where 080 unblocks one optional slice, and +090's own goal instruments "gateway event" telemetry, which does not exist until this slice +builds it. So the ROADMAP's own text implies 070 before 090. 070 also needs nothing of the +owner to start: every acceptance criterion runs against the in-process `Console` adapter, and the +real platforms are 071 and 072. + +**Correction in this commit.** `SLICE.md`'s milestone field read `M5 Always-on`, the name that was +split into M5a and M5b on 2026-09-08; slices 059 to 062 were corrected then and 070 was missed. +It now reads `M5b Reaches`, matching ROADMAP's milestone table. No scope changes with it. + +What this slice joins, measured today rather than remembered: sessions carry `origin` and +`origin_ref` already (010's migration), so the router needs no migration but its own +(`gateway_identities`); the stream is `Trinity.Sessions.Events`, topic `session:`, events +`{:user_message, _}`, `{:assistant_delta, _}`, `{:assistant_message, _}`, `{:tool_call, _}`, +`{:state, _}`, `{:turn_interrupted, _}`, `{:error, _}`, `{:compaction, _}`, `{:forked, _}`; +approvals broadcast on `approvals:` and `approvals:all` (021), and the only way to +decide one is `Trinity.Permissions.decide_request/3`; 050's delivery is a behaviour with +`deliver(Run.t(), Task.t())` and one implementation (`Delivery.Desktop`). `hammer` is **not** a +dependency and this slice does not add one: the rate limit is a token bucket in the router's own +state, so VERSIONS.md gains no row (AC6 measures the behaviour, not the library). + +The security model already decides the question this slice would otherwise have to ask: docs/07 +"Gateways" fixes the channel trust cap (`:read`, `:network`, `:write` approvable from a paired +gateway; `:exec` and `:destructive` desktop or console only, the cap applied **after** the gate's +own decision, never instead of it) and says a capped request is answered, not dropped. AC8 is that +paragraph as a test. + +## G1 plan, 2026-09-22 + +Branch `slice/070-gateway-core`; ROADMAP row 070 to `in_progress`. Each line names its test. + +1. `Trinity.Gateways` (a boundary whose deps follow docs/01's row and `Trinity.Scheduler`'s + shape: `[Trinity, Trinity.Sessions]`, with Permissions and PubSub reached through `Trinity`'s + exports), `Gateways.Adapter` (the behaviour: + `child_spec/1`, `capabilities/0`, `deliver/2`, `format/2`, `render_approval/2`) and + `Gateways.Console` (in-process adapter: a mailbox a test reads). Test: the behaviour's + callbacks are all implemented by Console, and `capabilities/0` drives chunking. +2. Migration `gateway_identities` (adapter, external_user_id, state, paired_at, allowlisted) with + `Gateways.Identities`; `Gateways.Pairing` (a one-time code, ten minutes, shown in the UI). + Tests: an unknown identity gets the pairing prompt and nothing else; a used or expired code is + refused (AC2's automatic half). +3. `Gateways.Router`: inbound tuple → identity check → rate limit → session lookup or create with + `origin` and `origin_ref` → `Sessions.send_user_message/2`; a per-conversation subscriber to + `session:` that coalesces `{:assistant_delta, _}` into an edit or a final message per the + adapter's capabilities. Tests: AC1 (streamed reply, `origin: "console"`), AC6 (the bucket), + AC7 (50 conversations under the fake provider, then `integrity_check`). +4. `Gateways.Commands` (a registry: `/new`, `/sessions`, `/model`, `/skills`, `/memory`, + `/approve`, `/deny`, `/attach`, `/help`). Tests: AC3 (`/attach` makes the desktop LiveView and + the console see one stream, both subscribed in the test), and each command's refusal path. +5. Approvals over chat: the router renders `approvals:` into the conversation and + `/approve` and `/deny` call `Permissions.decide_request/3`, **after** `Gateways.Cap` refuses a + tier the channel may not approve. Tests: AC4 (an approval mid-turn resumes on `/approve`) and + AC8 (a `:destructive` request is refused with the reason, the conversation is told where to + decide, and the refusal is receipted). +6. `Trinity.Scheduler.Delivery.Gateway` (050's behaviour, second implementation). Test: AC5, a + cron run delivered to a console conversation. +7. `/gateways` LiveView (adapter status, pairing codes, identities with allow and revoke) and + `mix trinity.console` for a dev REPL. Tests: the page lists, pairs and revokes. +8. Docs: docs/01 (the boundary and its modules), docs/07 (the Gateways section as built, the cap + as a property rather than a plan), README's "Not there yet" line; PROOF. + +Manual verification queue, so the owner sees it now and not at review: **AC2** (the pairing +prompt, then the code entered in the UI, as a screenshot) and **AC9** (the `/gateways` page). +Both are screenshots this machine can take with the cached headless chromium, as slices 013 to +060 did; neither needs an external account, a platform token or anything of the owner's. + +Not built here: the real platform adapters (071, 072), voice transcription, media beyond images, +and any change to the gate's own decision (the cap is applied after it, never instead of it). + +## Deviations from SLICE.md and the G1 plan (recorded before the commit that carries them) + +1. **The channel trust cap applies to `Console` too, and docs/07 is amended to say so.** That + paragraph read "`:exec` and `:destructive` are desktop or console only", written before there + was a console adapter. There is one now. Exempting it would make the in-process channel the one + channel that may approve a destructive effect, which is a hole shaped exactly like the thing + the cap exists for, and it would leave AC8 with nothing to test against. So `Cap` applies a + ceiling to every adapter, configurable per adapter, `:write` by default; docs/07 now reads + "desktop only" with an amendment note naming this slice. +2. **No `hammer`.** SLICE.md said "Hammer or a simple token bucket". The bucket is in the router's + own state, so the slice adds no dependency and `VERSIONS.md` gains no row. AC6 measures the + behaviour rather than the library. +3. **`/attach` binds through a pure state transition** (`Router.bind_state/4`), not through + `Router.attach/3`. Commands run inside the router process, so a command calling the router's + own API is that process calling itself: the suite found the deadlock the first time `/attach` + ran. `attach/3` stays for callers outside the process. +4. **`Trinity.Gateways` is a boundary under `Trinity`**, and the `/gateways` page reaches it + through `Trinity`'s exports, the way the tasks page reaches `Scheduler`. `TrinityWeb` could not + list it as a dependency directly: the library allows only a sibling, a parent, or an ancestor's + dependency, and the compiler said so. +5. **The adapter's name is derived, not declared** (`Adapter.name/1`: the last segment of the + module, underscored). A declared name a module can contradict is a second source of truth + (CLAUDE.md section 8). +6. **No adapter supervisor of its own.** The slice's tree listed `Trinity.Gateways.Supervisor`; + `Console` and `Router` are ordinary supervised children and a platform adapter will be one too, + so a supervisor whose only job is to hold one child was not written. docs/01's tree is updated + to what exists. + +## Findings + +- **F1.** `Commands` runs inside the router's process, so `/attach` calling `Router.attach/3` + deadlocked ("process attempted to call itself"). Fixed by making the binding a pure state + transition. Found by the AC3 test the first time it ran. +- **F2.** `/attach ` reached `Sessions.get_session/1` with something that is not a UUID, + and Ecto raised `Ecto.Query.CastError` **inside the router**, taking every conversation's + binding down with it. Two fixes, because the second is the one that matters: the prefix is + resolved without a cast, and `dispatch/2` now rescues, so a command that raises costs its own + message and nothing else. A test holds the router's pid across a malformed command. +- **F3.** `Console.text/1` renumbered message references per conversation while `deliver/2` hands + out a counter global to the process, so an edit landed on a different message and a later reply + could overwrite an earlier one. The reference is now recorded beside the message. Found by the + approvals suite, where a cap refusal overwrote the turn's reply. +- **F4.** `Format.plain/1` used `^\s*[-*]\s+` for a list marker; `\s` matches a newline, so the + blank line before a list was eaten and two paragraphs became one. `[ \t]` now. +- **F5.** Two approval tests asserted on "the last thing the channel was shown" while the opening + turn's reply was still in flight: they passed alone and failed when the suite ran whole. The + helper waits the opening turn out and reads everything shown rather than the last message. + +## Follow-ups + +- The real platforms (071, 072). This slice's `Console` is the shape they implement; nothing of + the router should need to change for them, and if one does, that is the finding to record. +- `Router` holds every conversation in one process. Fifty conversations are fine (AC7); a + deployment with thousands would want one process per conversation under a registry, which is a + slice of its own rather than a change to make on a guess. +- The `/gateways` page shows no per-identity message counts or last-seen time; 090's observability + is where that belongs. +- `Console.capabilities/0` carries `images: false`: an adapter that carries images needs a path + for them, which the SLICE's "Out" line already defers. diff --git a/slices/070-gateway-core/PROOF.md b/slices/070-gateway-core/PROOF.md new file mode 100644 index 0000000..f7efafb --- /dev/null +++ b/slices/070-gateway-core/PROOF.md @@ -0,0 +1,237 @@ +# Proof for slice 070: Gateway core (adapter behaviour, routing, PubSub fan-out) + +Agent: Claude Opus 5 (1M context) · Date: 2026-09-22 · Branch: slice/070-gateway-core · Final commit: written at the close, below + +## Summary + +A channel can reach Trinity now: `Trinity.Gateways.Adapter` is the behaviour a platform +implements, `Router` is the one road from a channel to a session, and `Console` is an in-process +adapter that ships, so every criterion here runs without an account, a token or a network. An +unknown sender gets a pairing code and nothing else; a paired one gets slash commands, a streamed +reply, and any approval the turn raises rendered into the same conversation. What a channel may +approve is capped below what the desktop may, and a capped answer is refused in words and +receipted. What was hard: three bugs the suite found rather than review (NOTES, "Findings"), one +of them a self-call deadlock and one a reference mix-up in the console's own log. Deferred: the +real platforms are 071 and 072, as the slice says. + +## Gate + +``` +$ mix gate +prod_check: PASS +3095 mods/funs, found no issues. +... SCAN COMPLETE ... +Result: 564 passed, 18 excluded +trinity.coverage: 050 80.31% vs 061 80.05%: OK +plan_check: PASS +gate exit 0 +``` + +## Tests + +``` +$ mix test --cover +Finished in 109.2 seconds (0.7s async, 108.4s sync) +Result: 564 passed, 18 excluded +| 80.89% | Total | +``` + +This slice's modules: + +``` +| 100.00% | Trinity.Gateways | +| 100.00% | Trinity.Gateways.Adapter | +| 100.00% | Trinity.Gateways.Console | +| 96.77% | Trinity.Gateways.Format | +| 95.59% | TrinityWeb.GatewaysLive | +| 92.86% | Trinity.Gateways.Cap | +| 92.00% | Trinity.Gateways.Identities | +| 90.00% | Trinity.Gateways.Receipts | +| 84.00% | Trinity.Gateways.Commands | +| 70.59% | Trinity.Gateways.Router | +| 40.00% | Trinity.Gateways.Identity | +``` + +`Identity` is a schema: its uncovered lines are the changeset's unexercised validation branches. +`Router`'s uncovered lines are the paths a platform adapter reaches and the console does not: the +no-edit branch of the stream (`Console` can edit) and the `{:error, _}` broadcast. + +The slice's own suites, named: + +``` +$ mix test test/trinity/gateways test/trinity_web/live/gateways_live_test.exs --trace +Result: 39 passed +``` + +## Acceptance criteria evidence + +### AC1 [auto]: Console adapter: inbound from a paired identity creates a session with `origin: "console"` and returns a streamed reply + +``` +* test AC1: a paired identity's message creates a session with the adapter's origin, and the reply streams back (29.6ms) +``` + +`test/trinity/gateways/router_test.exs`. The message is placed, the reply arrives in the channel, +and the session's `origin` is `"console"` with `origin_ref` `%{"adapter", "conversation", +"external_user_id"}`. The user's message is in the session's history, so the channel and the +desktop are looking at one conversation and not two. + +### AC2 [manual + auto]: an unpaired identity receives only a pairing prompt; after the code shown in the UI, the next message is processed + +Automatic half: + +``` +* test AC2 (auto): an unknown sender is shown a code and nothing else happens (1.4ms) +* test an unknown sender is pending with a live code, and nothing else is created (0.3ms) +* test the code pairs the sender once; a wrong code, an expired one and a second use are refused (0.4ms) +* test an expired code is refused, and the next message is given a fresh one (1.0ms) +* test a revoked identity cannot pair and is not admitted (0.5ms) +* test an allowlisted id is paired at first sight, with no code shown (0.3ms) +* test the same external id on two adapters is two identities (0.6ms) +``` + +"Nothing else" is asserted and not asserted loosely: no session exists (`Sessions.list_sessions() +== []`) and the provider was never called (`Fake.calls() == 0`) after an unknown sender writes. + +Manual half, from a running application (`scripts/dev_gateways.sh`, the full transcript in +`proof/ac2-pairing-transcript.txt`): + +``` +=== AC2: an unknown sender is answered with a pairing code and nothing else === + they say: are my tasks done? + result: {:error, :pending} + trinity: Trinity does not know you yet. Open the desktop app, go to Settings → Gateways, and you will + see the code HEFAXD. Send that code here to pair. It lasts 10 minutes. + sessions created so far: 0 (a stranger gets no session) +=== the wrong code is refused, and says only that === + they say: AAAAAA + result: {:error, :pending} +=== the code from /gateways pairs them === + they say: HEFAXD + result: {:ok, :paired} + trinity: Paired. Say anything and Trinity will answer. +=== and the next message is a message === + they say: are my tasks done? + result: {:ok, :placed} + trinity: Yes: three tasks are due today, and the notes directory changed. + sessions now: 1 +``` + +The code as the owner sees it: `proof/ac9-gateways-page.png` (the "Waiting to pair" row). That +screenshot is the other half of this criterion as well as AC9's: the code is shown on that page +and nowhere else, which is what makes reading it the proof of pairing. + +### AC3 [auto]: `/attach` puts a console conversation and a desktop session on one stream + +``` +* test AC3: /attach puts the conversation and the desktop on one session (1058.9ms) +``` + +Both directions are asserted in one test: a message sent through `Sessions.send_user_message/2` +as the LiveView sends it appears in the console's stream, and a message from the console lands in +that same session (`Sessions.list_sessions()` still holds one). + +### AC4 [auto]: an approval raised during a gateway turn is rendered in the console; `/approve` resumes it + +``` +* test AC4: a request raised in a gateway session is rendered there and /approve decides it (56.4ms) +* test /deny from the channel denies it, and an unknown id says so (56.9ms) +* test a raised request in someone else's session is not rendered into this channel (129.1ms) +``` + +The decision goes through `Permissions.decide_request/3`, so the row records +`decided_by: "gateway:console:u-1"`: who answered is the channel and the account, never "the UI". +A request raised in another session is not rendered here, which is the other half of "an approval +belongs to the surface that asked for it". + +### AC5 [auto]: cron delivery to a gateway conversation + +``` +* test AC5: the run's summary reaches the conversation, and the run is marked delivered (1.6ms) +* test a summary longer than the channel's limit arrives in chunks, not truncated (0.9ms) +* test an unknown adapter or a missing conversation is an error, and nothing is marked delivered (19.7ms) +``` + +`Trinity.Scheduler.Delivery.Gateway` is 050's behaviour with a second implementation, resolved +from the task row by `Delivery.for/1`. The row names an adapter, never a module: a module name in +a row the scheduler would call is a way to run any module by writing a row. + +### AC6 [auto]: past N messages a minute an identity is throttled + +``` +* test AC6: past its rate limit an identity is answered, not dropped, and recovers (2.1ms) +``` + +A token bucket in the router's own state, above the session lookup, so a flood costs a row lookup +and nothing more. Each identity has its own bucket, and a throttled sender is told rather than +ignored. No new dependency: `hammer` is not in the tree and this slice does not add one. + +### AC7 [auto]: 50 console conversations complete; `integrity_check` ok + +``` +* test AC7: fifty conversations complete, each in its own session, and the database is intact (210.9ms) +``` + +Fifty paired identities talk at once through one router process. Every message is placed, every +conversation is answered **in its own channel** (`refute text =~ "hello from"` catches a stream +leaking into another conversation), there are fifty distinct sessions, and `PRAGMA +integrity_check` is `ok` (asked only of SQLite, by `Trinity.Repo.__adapter__()`). + +### AC8 [auto]: a `:destructive` approval from a gateway is refused under the default cap, the conversation is told where to decide, and the refusal is receipted + +``` +* test AC8: a destructive request is refused by the cap, receipted, and left for the desktop (64.2ms) +* test the ceiling is per adapter and configurable, and an unknown tier is refused (0.9ms) +``` + +Four things are asserted, and the third is the one that matters: the rendered request already +says the channel cannot answer it; `/approve` is refused in words; **the gate is never asked** +(`%Approval{status: "pending", decided_by: nil}` afterwards, so the request is still there for +the desktop); and the refusal is a decision receipt on the session's own chain with basis +`channel_cap`, naming the adapter, the account and the tier. An unknown tier is refused rather +than waved through. + +### AC9 [manual]: the `/gateways` page + +`proof/ac9-gateways-page.png`, from the running application. It shows the channel with the tier it +may approve (`console … approves up to write`), a live pairing code with allow and revoke, and the +identities list with a paired and a pending row. Automatic half: + +``` +* test the page lists the channel, whether it runs, and the tier it may approve (2.8ms) +* test a waiting identity's code is shown here, and allow pairs it without a code (3.6ms) +* test revoking keeps the row and shows it as revoked (89.8ms) +* test with no identity and no adapter configured the page says so (1.9ms) +``` + +## Manual verification for the reviewer + +Both `[manual]` criteria have an artefact in `proof/` above. To see either again: + +``` +$ mix assets.build && ./scripts/dev_gateways.sh # prints the transcript, then PORT= +``` + +Then open `http://127.0.0.1:/gateways`. The script runs on `trinity_screenshots.db`, never the +suite's. Kill it by the pid `pgrep -af beam.smp` prints, never by a pattern that also matches your +own shell. + +For the channel side by hand rather than scripted: `mix trinity.console` talks to the same router +from a terminal, and pairs the same way. + +## Deviations from SLICE.md + +See NOTES.md, "Deviations". In short: the cap applies to `Console` like every other adapter +(docs/07 said "desktop or console only" before a console adapter existed, and is amended in this +slice); the rate limit is a token bucket rather than Hammer, so no dependency is added; and +`/attach`'s binding is a pure state transition because a command runs inside the router process. + +## Versions touched + +`VERSIONS.md` unchanged: this slice adds no dependency. + +## Git + +``` +$ git log --oneline main..HEAD +``` diff --git a/slices/070-gateway-core/SLICE.md b/slices/070-gateway-core/SLICE.md index 5de1896..d08c069 100644 --- a/slices/070-gateway-core/SLICE.md +++ b/slices/070-gateway-core/SLICE.md @@ -3,7 +3,7 @@ | Field | Value | |---|---| | Phase | 7 Gateways | -| Milestone | M5 Always-on | +| Milestone | M5b Reaches | | Size | M | | Depends on | 012 | diff --git a/slices/070-gateway-core/proof/ac2-pairing-transcript.txt b/slices/070-gateway-core/proof/ac2-pairing-transcript.txt new file mode 100644 index 0000000..e4c9810 --- /dev/null +++ b/slices/070-gateway-core/proof/ac2-pairing-transcript.txt @@ -0,0 +1,35 @@ +# AC2, the pairing flow, from scripts/dev_gateways.sh on 2026-09-22 +# The console adapter through Trinity.Gateways.Router: the same path a platform adapter takes. + +=== AC2: an unknown sender is answered with a pairing code and nothing else === + + they say: are my tasks done? + result: {:error, :pending} + trinity: Trinity does not know you yet. Open the desktop app, go to Settings → Gateways, and you will + see the code HEFAXD. Send that code here to pair. It lasts 10 minutes. + + sessions created so far: 0 (a stranger gets no session) + the page shows the code: HEFAXD + +=== the wrong code is refused, and says only that === + + they say: AAAAAA + result: {:error, :pending} + trinity: Trinity does not know you yet. Open the desktop app, go to Settings → Gateways, and you will + see the code HEFAXD. Send that code here to pair. It lasts 10 minutes. + +=== the code from /gateways pairs them === + + they say: HEFAXD + result: {:ok, :paired} + trinity: Paired. Say anything and Trinity will answer. + +=== and the next message is a message === + + they say: are my tasks done? + result: {:ok, :placed} + trinity: Yes: three tasks are due today, and the notes directory changed. + + sessions now: 1 + +PORT=44049 diff --git a/slices/070-gateway-core/proof/ac9-gateways-page.png b/slices/070-gateway-core/proof/ac9-gateways-page.png new file mode 100644 index 0000000..dc42f3d Binary files /dev/null and b/slices/070-gateway-core/proof/ac9-gateways-page.png differ diff --git a/test/trinity/gateways/approvals_test.exs b/test/trinity/gateways/approvals_test.exs new file mode 100644 index 0000000..930132d --- /dev/null +++ b/test/trinity/gateways/approvals_test.exs @@ -0,0 +1,175 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.ApprovalsTest do + @moduledoc """ + Slice 070, AC4 and AC8: an approval raised by a turn that started in a channel is rendered into + that channel and answered there with `/approve`; and a request above the channel's tier ceiling + is refused in words, receipted, and left pending for the desktop, with the gate never asked. + The cap is applied after the gate's own decision and never instead of it (docs/07). + """ + use Trinity.SessionCase + + alias Trinity.Gateways.{Cap, Console, Identities, Router} + alias Trinity.LLM.Providers.Fake + alias Trinity.Permissions + alias Trinity.Permissions.Approval + alias Trinity.Receipts + + @adapter Console + @conv "c-approve" + @user "u-1" + + setup do + start_supervised!(Console) + start_supervised!(Router) + {:ok, identity, :pending} = Identities.admit("console", @user) + {:ok, _} = Identities.pair("console", @user, identity.code) + :ok + end + + # The opening turn is waited out here, not just started: its reply lands asynchronously, and a + # later assertion on "the last thing the channel was shown" is otherwise racing it. Two tests + # failed exactly that way when the suite ran as a whole and passed when run alone. + defp session_for_conversation do + Fake.scripts([script_deltas(1, "hi ")]) + {:ok, :placed} = Router.inbound(@adapter, @conv, @user, "start a session") + session_id = await(fn -> Router.session_of(@adapter, @conv) end) + await_shown("hi") + session_id + end + + defp await(fun, timeout \\ 5_000) do + deadline = System.monotonic_time(:millisecond) + timeout + do_await(fun, deadline) + end + + defp do_await(fun, deadline) do + case fun.() do + nil -> + if System.monotonic_time(:millisecond) > deadline, + do: flunk("never became truthy"), + else: Process.sleep(25) && do_await(fun, deadline) + + value -> + value + end + end + + # Everything the channel has been shown, joined. "" and not nil when there is nothing yet: + # `nil =~ "x"` raises, which turns a wait into an immediate failure and hides the race it was + # written to wait out. A wait looks here and not at the last message alone: + # the turn's reply and an approval raised during it arrive in either order, and a wait that + # reads only the last one waits forever for whichever came first. + defp shown, do: Console.text(@conv) |> Enum.join("\n") + + defp await_shown(fragment), do: await(fn -> if shown() =~ fragment, do: :shown end) + + test "AC4: a request raised in a gateway session is rendered there and /approve decides it" do + session_id = session_for_conversation() + + {:ok, %Approval{id: id}} = + Permissions.request_approval(session_id, "write_note", %{"path" => "notes/a.md"}, + risk: :write + ) + + await_shown("Approval needed: write_note") + assert shown() =~ "/approve #{String.slice(id, 0, 8)}" + + assert {:ok, :command} = + Router.inbound(@adapter, @conv, @user, "/approve #{String.slice(id, 0, 8)}") + + assert shown() =~ "Approved #{String.slice(id, 0, 8)}" + + approval = Permissions.get_approval(id) + assert approval.status == "allowed" and approval.decision == "once" + assert approval.decided_by == "gateway:console:#{@user}" + end + + test "/deny from the channel denies it, and an unknown id says so" do + session_id = session_for_conversation() + + {:ok, %Approval{id: id}} = + Permissions.request_approval(session_id, "fetch", %{"url" => "https://example.com"}, + risk: :network + ) + + await_shown("Approval needed: fetch") + + assert {:ok, :command} = + Router.inbound(@adapter, @conv, @user, "/deny #{String.slice(id, 0, 8)}") + + assert shown() =~ "Denied" + assert %Approval{status: "denied"} = Permissions.get_approval(id) + + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/approve zzzzzzzz") + assert shown() =~ "No pending approval starts with zzzzzzzz" + end + + test "AC8: a destructive request is refused by the cap, receipted, and left for the desktop" do + session_id = session_for_conversation() + assert Cap.ceiling(@adapter) == :write + refute Cap.allows?(@adapter, :destructive) + + {:ok, %Approval{id: id}} = + Permissions.request_approval(session_id, "delete_tree", %{"path" => "/tmp/x"}, + risk: :destructive + ) + + # It is rendered, and the render already says the channel cannot answer it. + await_shown("Approval needed: delete_tree") + assert shown() =~ "may approve up to write" + assert shown() =~ "permissions page" + + # And trying anyway is refused in words rather than dropped. + assert {:ok, :command} = + Router.inbound(@adapter, @conv, @user, "/approve #{String.slice(id, 0, 8)}") + + assert shown() =~ "That is a destructive request" + + # The gate was never asked: the request is still pending for the desktop. + assert %Approval{status: "pending", decided_by: nil} = Permissions.get_approval(id) + + # The refusal is a receipt on the session's chain, naming the channel and the account. + scope = Receipts.session_scope(session_id) + + refusal = + await(fn -> + Enum.find(Receipts.list(scope), &(&1.subject["phase"] == "gateway_cap")) + end) + + assert refusal.subject["adapter"] == "console" + assert refusal.subject["external_user_id"] == @user + assert refusal.subject["risk"] == "destructive" + assert Jason.decode!(refusal.signed_payload)["decision"]["basis"] == "channel_cap" + on_exit(fn -> Receipts.stop_writer(scope) end) + end + + test "the ceiling is per adapter and configurable, and an unknown tier is refused" do + assert Cap.allows?(@adapter, :read) + assert Cap.allows?(@adapter, "write") + refute Cap.allows?(@adapter, :exec) + + Application.put_env(:trinity, :gateways, caps: %{"console" => :destructive}) + on_exit(fn -> Application.delete_env(:trinity, :gateways) end) + assert Cap.allows?(@adapter, :destructive) + + # A tier this module does not know is not a tier it waves through. + refute Cap.allows?(@adapter, :root) + refute Cap.allows?(@adapter, nil) + end + + test "a raised request in someone else's session is not rendered into this channel" do + _mine = session_for_conversation() + + {:ok, other} = + Sessions.create_session(%{persona_id: Sessions.default_persona().id, title: "desktop"}) + + # Wait for this conversation's own turn to finish first, or "nothing new arrived" is only a + # statement about how fast the assertion ran. + before = Console.text(@conv) + + {:ok, _} = Permissions.request_approval(other.id, "write_note", %{}, risk: :write) + Process.sleep(100) + assert Console.text(@conv) == before + end +end diff --git a/test/trinity/gateways/console_test.exs b/test/trinity/gateways/console_test.exs new file mode 100644 index 0000000..756cad1 --- /dev/null +++ b/test/trinity/gateways/console_test.exs @@ -0,0 +1,78 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.ConsoleTest do + @moduledoc """ + Slice 070: the in-process adapter. It implements every callback of the behaviour, it records + what it was asked to deliver, and `text/1` answers with what a person in the channel would be + looking at: a message replaced by its edits rather than a log of both. + """ + use ExUnit.Case, async: false + + alias Trinity.Gateways.{Adapter, Console} + + setup do + pid = start_supervised!(Console) + on_exit(fn -> if Process.alive?(pid), do: :ok end) + :ok + end + + test "it implements every callback the behaviour declares" do + callbacks = Adapter.behaviour_info(:callbacks) |> Enum.sort() + exported = Console.__info__(:functions) + + for {name, arity} <- callbacks do + assert {name, arity} in exported or name == :child_spec, + "Console does not implement #{name}/#{arity}" + end + + # child_spec/1 comes from `use GenServer`, which is how the supervisor starts it. + assert function_exported?(Console, :child_spec, 1) + end + + # Written as one equality and not as `assert caps.markdown`: the type checker inlines the + # module attribute and flags an assertion on a literal, rightly, since it proves nothing. What + # is worth holding is the whole declaration, so a change to any field is a change to this line. + test "capabilities declare a plain text channel that can edit, and every field the router reads" do + assert Console.capabilities() == %{ + markdown: true, + images: false, + buttons: false, + edits: true, + max_length: 4_000 + } + end + + test "a delivered message answers with a reference, and an edit replaces what it names" do + assert {:ok, ref} = Console.deliver("c1", {:message, "first"}) + assert :ok = Console.deliver("c1", {:typing, true}) + assert :ok = Console.deliver("c1", {:edit, ref, "first, corrected"}) + assert {:ok, _} = Console.deliver("c1", {:message, "second"}) + + assert Console.text("c1") == ["first, corrected", "second"] + + assert Console.delivered("c1") == [ + {:message, "first"}, + {:typing, true}, + {:edit, ref, "first, corrected"}, + {:message, "second"} + ] + end + + test "conversations are separate, and one can be cleared" do + Console.deliver("a", {:message, "for a"}) + Console.deliver("b", {:message, "for b"}) + assert Console.text("a") == ["for a"] + assert Console.text("b") == ["for b"] + + assert :ok = Console.clear("a") + assert Console.text("a") == [] + assert Console.text("b") == ["for b"] + end + + test "format and render_approval are the shared defaults, not a dialect of its own" do + long = String.duplicate("word ", 2_000) + chunks = Console.format(long, Console.capabilities()) + assert Enum.all?(chunks, &(String.length(&1) <= Console.capabilities().max_length)) + assert length(chunks) > 1 + end +end diff --git a/test/trinity/gateways/delivery_test.exs b/test/trinity/gateways/delivery_test.exs new file mode 100644 index 0000000..21d79ca --- /dev/null +++ b/test/trinity/gateways/delivery_test.exs @@ -0,0 +1,74 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.DeliveryTest do + @moduledoc """ + Slice 070 AC5: a scheduled run delivered to a channel. The summary reaches the conversation + through the adapter's own formatting, the run is marked delivered and broadcast as the desktop + delivery does, and a destination naming an adapter that is not configured is an error rather + than a silent drop. + """ + use Trinity.DataCase, async: false + + alias Trinity.Gateways.Console + alias Trinity.Scheduler + alias Trinity.Scheduler.Delivery + + setup do + start_supervised!(Console) + Application.put_env(:trinity, :gateways, adapters: [Console]) + on_exit(fn -> Application.delete_env(:trinity, :gateways) end) + :ok + end + + defp task_to(destination) do + {:ok, task} = + Scheduler.create_task(%{ + name: "morning digest", + kind: "cron", + schedule: "0 9 * * *", + prompt: "summarise", + deliver_to: destination + }) + + task + end + + test "AC5: the run's summary reaches the conversation, and the run is marked delivered" do + task = task_to(%{"kind" => "gateway", "adapter" => "console", "conversation" => "c-cron"}) + {:ok, run} = Scheduler.run_now(task) + {:ok, run} = Scheduler.update_run(run, %{status: "ok", summary: "three things happened"}) + + # The scheduler picks the implementation from the row, not from a module in it. + assert Delivery.for(task) == Delivery.Gateway + assert {:ok, delivered} = Delivery.for(task).deliver(run, task) + + assert Console.text("c-cron") == ["morning digest: three things happened"] + assert delivered.delivered_at != nil + assert Scheduler.get_run(run.id).delivered_at != nil + end + + test "a summary longer than the channel's limit arrives in chunks, not truncated" do + task = task_to(%{"kind" => "gateway", "adapter" => "console", "conversation" => "c-long"}) + {:ok, run} = Scheduler.run_now(task) + summary = String.duplicate("word ", 1_500) + {:ok, run} = Scheduler.update_run(run, %{status: "ok", summary: summary}) + + assert {:ok, _} = Delivery.Gateway.deliver(run, task) + chunks = Console.text("c-long") + assert length(chunks) > 1 + assert Enum.all?(chunks, &(String.length(&1) <= Console.capabilities().max_length)) + assert chunks |> Enum.join(" ") =~ "word word" + end + + test "an unknown adapter or a missing conversation is an error, and nothing is marked delivered" do + task = task_to(%{"kind" => "gateway", "adapter" => "telegram", "conversation" => "c-1"}) + {:ok, run} = Scheduler.run_now(task) + {:ok, run} = Scheduler.update_run(run, %{status: "ok", summary: "x"}) + + assert {:error, {:unknown_adapter, "telegram"}} = Delivery.Gateway.deliver(run, task) + assert Scheduler.get_run(run.id).delivered_at == nil + + incomplete = task_to(%{"kind" => "gateway", "adapter" => "console"}) + assert {:error, {:missing, "conversation"}} = Delivery.Gateway.deliver(run, incomplete) + end +end diff --git a/test/trinity/gateways/format_test.exs b/test/trinity/gateways/format_test.exs new file mode 100644 index 0000000..7e64e7a --- /dev/null +++ b/test/trinity/gateways/format_test.exs @@ -0,0 +1,84 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.FormatTest do + @moduledoc """ + Slice 070: the default formatting every adapter delegates to. Chunking splits at the latest + boundary that fits and never past the limit; a split code fence is closed and re-opened; a + channel without markdown is given the markers removed rather than shown; an approval renders + with the two commands a person can type. + """ + use ExUnit.Case, async: true + + alias Trinity.Gateways.Format + alias Trinity.Permissions.Approval + + @caps %{markdown: true, images: false, buttons: false, edits: true, max_length: 40} + + test "a message inside the limit is one chunk, unchanged" do + assert Format.chunk("short enough", 40) == ["short enough"] + assert Format.format("short enough", @caps) == ["short enough"] + end + + test "chunks never exceed the limit and prefer the latest paragraph, line, then word boundary" do + paragraphs = "#{String.duplicate("a", 20)}\n\n#{String.duplicate("b", 20)}" + assert [first, second] = Format.chunk(paragraphs, 30) + assert first == String.duplicate("a", 20) + assert second == String.duplicate("b", 20) + + lines = "#{String.duplicate("a", 20)}\n#{String.duplicate("b", 20)}" + assert [^first, ^second] = Format.chunk(lines, 30) + + words = String.duplicate("word ", 20) + chunks = Format.chunk(words, 30) + assert Enum.all?(chunks, &(String.length(&1) <= 30)) + + assert chunks |> Enum.join(" ") |> String.replace(~r/\s+/, " ") |> String.trim() == + String.trim(words) + end + + test "a single word longer than the limit is cut, because there is no boundary to prefer" do + long = String.duplicate("x", 95) + chunks = Format.chunk(long, 40) + assert Enum.map(chunks, &String.length/1) == [40, 40, 15] + assert Enum.join(chunks) == long + end + + test "a code fence split across chunks is closed and re-opened, so the rest is not swallowed" do + code = "```\n" <> String.duplicate("line of code\n", 8) <> "```" + chunks = Format.chunk(code, 60) + assert length(chunks) > 1 + assert Enum.all?(chunks, &(String.length(&1) <= 60 + 8)) + + # Every chunk holds an even number of fences: none of them leaves a block open. + for chunk <- chunks do + assert chunk |> String.split("```") |> length() |> rem(2) == 1, + "unbalanced fence in #{chunk}" + end + end + + test "a channel without markdown is given the markers removed, not shown" do + caps = %{@caps | markdown: false, max_length: 4_000} + text = "# Heading\n\nSome **bold** and *italic* and `code`.\n\n- one\n- two" + assert [plain] = Format.format(text, caps) + assert plain == "Heading\n\nSome bold and italic and code.\n\n• one\n• two" + refute plain =~ "**" + refute plain =~ "#" + end + + test "an approval renders with its tool, its tier and the two commands that answer it" do + approval = %Approval{ + id: "0198f0c0-1111-2222-3333-444455556666", + tool: "write_note", + risk: "write", + args: %{"path" => "notes/today.md", "body" => String.duplicate("x", 80)} + } + + assert {:message, text} = Format.render_approval(approval, %{@caps | max_length: 4_000}) + assert text =~ "Approval needed: write_note (write)" + assert text =~ "path=notes/today.md" + assert text =~ "/approve 0198f0c0" + assert text =~ "/deny 0198f0c0" + # The arguments are summarised, never pasted whole. + refute text =~ String.duplicate("x", 45) + end +end diff --git a/test/trinity/gateways/identities_test.exs b/test/trinity/gateways/identities_test.exs new file mode 100644 index 0000000..526cb0b --- /dev/null +++ b/test/trinity/gateways/identities_test.exs @@ -0,0 +1,95 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.IdentitiesTest do + @moduledoc """ + Slice 070, AC2's automatic half and the rule docs/07 states: an unknown sender is admitted as + `pending` with a code and nothing else; the code pairs them once, is refused when wrong, + expired or already used; a revoked identity cannot pair; an allowlisted id is paired at first + sight; and a pending identity whose code has expired is given a new one rather than being + locked out. + """ + use Trinity.DataCase, async: false + + alias Trinity.Gateways.{Identities, Identity} + + @adapter "console" + @user "u-1" + + test "an unknown sender is pending with a live code, and nothing else is created" do + assert {:ok, %Identity{} = identity, :pending} = + Identities.admit(@adapter, @user, conversation: "c1", display_name: "Someone") + + assert identity.state == "pending" + assert String.length(identity.code) == 6 + assert identity.code =~ ~r/^[A-Z2-9]+$/ + assert DateTime.compare(identity.code_expires_at, DateTime.utc_now()) == :gt + assert identity.paired_at == nil + assert identity.last_conversation == "c1" + assert Identities.pending() |> Enum.map(& &1.id) == [identity.id] + end + + test "the code pairs the sender once; a wrong code, an expired one and a second use are refused" do + {:ok, identity, :pending} = Identities.admit(@adapter, @user) + + assert {:error, :wrong_code} = Identities.pair(@adapter, @user, "AAAAAA") + assert {:ok, paired} = Identities.pair(@adapter, @user, String.downcase(identity.code)) + assert paired.state == "paired" and paired.paired_at != nil + assert paired.code == nil, "the code is spent, not kept" + + # A second attempt with the same code has nothing to match. + assert {:error, :no_code} = Identities.pair(@adapter, @user, identity.code) + + # And the next message is simply admitted. + assert {:ok, _, :paired} = Identities.admit(@adapter, @user) + end + + test "an expired code is refused, and the next message is given a fresh one" do + {:ok, identity, :pending} = Identities.admit(@adapter, @user) + stale = DateTime.add(DateTime.utc_now(), -1, :second) + {:ok, _} = identity |> Identity.changeset(%{code_expires_at: stale}) |> Trinity.Repo.update() + + assert {:error, :expired} = Identities.pair(@adapter, @user, identity.code) + assert Identities.pending() == [] + + assert {:ok, refreshed, :pending} = Identities.admit(@adapter, @user) + assert refreshed.code != identity.code + assert {:ok, _} = Identities.pair(@adapter, @user, refreshed.code) + end + + test "a revoked identity cannot pair and is not admitted" do + {:ok, identity, :pending} = Identities.admit(@adapter, @user) + assert {:ok, revoked} = Identities.revoke(identity) + assert revoked.state == "revoked" and revoked.revoked_at != nil + + assert {:ok, _, :revoked} = Identities.admit(@adapter, @user) + assert {:error, :revoked} = Identities.pair(@adapter, @user, identity.code) + # The row stays: a refusal is visible on the page rather than absent from it. + assert Enum.map(Identities.list(), & &1.id) == [identity.id] + end + + test "an allowlisted id is paired at first sight, with no code shown" do + Application.put_env(:trinity, :gateways, allowlist: [{"console", "trusted"}]) + on_exit(fn -> Application.delete_env(:trinity, :gateways) end) + + assert {:ok, identity, :paired} = Identities.admit(@adapter, "trusted") + assert identity.state == "paired" and identity.code == nil + assert Identities.pending() == [] + + # The allowlist names one pair, not a platform: another id on the same adapter still pairs. + assert {:ok, _, :pending} = Identities.admit(@adapter, "someone-else") + end + + test "the same external id on two adapters is two identities" do + assert {:ok, one, :pending} = Identities.admit("console", "same-id") + assert {:ok, two, :pending} = Identities.admit("other", "same-id") + assert one.id != two.id + assert {:ok, _} = Identities.pair("console", "same-id", one.code) + assert {:ok, _, :pending} = Identities.admit("other", "same-id") + end + + test "the desktop can pair and revoke without a code" do + {:ok, identity, :pending} = Identities.admit(@adapter, @user) + assert {:ok, allowed} = Identities.allow(identity) + assert allowed.state == "paired" and allowed.code == nil + end +end diff --git a/test/trinity/gateways/router_test.exs b/test/trinity/gateways/router_test.exs new file mode 100644 index 0000000..50019c4 --- /dev/null +++ b/test/trinity/gateways/router_test.exs @@ -0,0 +1,192 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.RouterTest do + @moduledoc """ + Slice 070 through the router, against the console adapter and the fake provider: + AC1 (a paired identity's message creates a session with the adapter's origin and the reply is + streamed back), AC2's automatic half (an unknown sender is answered with the code and nothing + else: no session exists for them), AC3 (`/attach` puts a conversation and the desktop on one + session), AC6 (the rate limit answers rather than dropping), and the commands. + """ + use Trinity.SessionCase + + alias Trinity.Gateways.{Console, Identities, Router} + alias Trinity.LLM.Providers.Fake + + @adapter Console + @conv "c-1" + @user "u-1" + + setup do + start_supervised!(Console) + start_supervised!(Router) + :ok + end + + defp pair!(user \\ @user) do + {:ok, identity, :pending} = Identities.admit("console", user) + {:ok, _} = Identities.pair("console", user, identity.code) + :ok + end + + defp await_text(conversation, expected, timeout \\ 5_000) do + deadline = System.monotonic_time(:millisecond) + timeout + do_await(conversation, expected, deadline) + end + + defp do_await(conversation, expected, deadline) do + text = Console.text(conversation) |> Enum.join("\n") + + cond do + text =~ expected -> + text + + System.monotonic_time(:millisecond) > deadline -> + flunk("never saw #{inspect(expected)} in #{inspect(text)}") + + true -> + Process.sleep(25) && do_await(conversation, expected, deadline) + end + end + + test "AC2 (auto): an unknown sender is shown a code and nothing else happens" do + assert {:error, :pending} = Router.inbound(@adapter, @conv, @user, "hello?") + + [shown] = Console.text(@conv) + {:ok, identity, _} = Identities.admit("console", @user) + assert shown =~ identity.code + assert shown =~ "pair" + + # No session was created and the model was never called for someone not let in. + assert Router.session_of(@adapter, @conv) == nil + assert Sessions.list_sessions() == [] + assert Fake.calls() == 0 + + # The code pairs them, and only then does a message become a message. + assert {:ok, :paired} = Router.inbound(@adapter, @conv, @user, identity.code) + assert Console.text(@conv) |> List.last() =~ "Paired" + end + + test "AC1: a paired identity's message creates a session with the adapter's origin, and the reply streams back" do + :ok = pair!() + Fake.scripts([script_deltas(3, "hello ")]) + + assert {:ok, :placed} = Router.inbound(@adapter, @conv, @user, "say hello") + assert await_text(@conv, "hello hello hello") + + session_id = Router.session_of(@adapter, @conv) + assert is_binary(session_id) + session = Sessions.get_session(session_id) + assert session.origin == "console" + + assert session.origin_ref == %{ + "adapter" => "console", + "conversation" => @conv, + "external_user_id" => @user + } + + # The user's message is in the session's history, so the channel and the desktop agree. + assert Enum.any?( + Sessions.history(session_id), + &(&1.role == "user" and &1.content == "say hello") + ) + end + + test "AC3: /attach puts the conversation and the desktop on one session" do + :ok = pair!() + + {:ok, desktop} = + Sessions.create_session(%{persona_id: Sessions.default_persona().id, title: "desktop"}) + + start_drained(desktop.id) + + assert {:ok, :command} = + Router.inbound(@adapter, @conv, @user, "/attach #{String.slice(desktop.id, 0, 8)}") + + assert Console.text(@conv) |> List.last() =~ "Attached to #{String.slice(desktop.id, 0, 8)}" + assert Router.session_of(@adapter, @conv) == desktop.id + + # A message sent from the desktop (as the LiveView does) reaches the console's stream. + Fake.scripts([script_deltas(2, "shared ")]) + {:ok, _} = Sessions.send_user_message(desktop.id, "from the desktop") + assert await_text(@conv, "shared shared") + + # And a message from the console lands in the same session, not a new one. + Fake.scripts([script_deltas(1, "back ")]) + assert {:ok, :placed} = Router.inbound(@adapter, @conv, @user, "from the channel") + assert await_text(@conv, "back") + assert Router.session_of(@adapter, @conv) == desktop.id + assert length(Sessions.list_sessions()) == 1 + end + + test "AC6: past its rate limit an identity is answered, not dropped, and recovers" do + :ok = pair!() + Application.put_env(:trinity, :gateways, rate_limit: [capacity: 2, per_minute: 60]) + on_exit(fn -> Application.delete_env(:trinity, :gateways) end) + Fake.scripts([script_deltas(1, "ok ")]) + + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/help") + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/help") + assert {:error, :rate_limited} = Router.inbound(@adapter, @conv, @user, "/help") + assert Console.text(@conv) |> List.last() =~ "Too many messages" + + # A different identity has its own bucket. + :ok = pair!("u-2") + assert {:ok, :command} = Router.inbound(@adapter, "c-2", "u-2", "/help") + end + + test "the commands answer without calling the model" do + :ok = pair!() + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/help") + assert Console.text(@conv) |> List.last() =~ "/attach" + + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/nope") + assert Console.text(@conv) |> List.last() =~ "I do not know /nope" + + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/sessions") + assert Console.text(@conv) |> List.last() =~ "No sessions yet" + + assert Fake.calls() == 0 + end + + test "/new unbinds the conversation, so the next message starts a session of its own" do + :ok = pair!() + Fake.scripts([script_deltas(1, "one ")]) + assert {:ok, :placed} = Router.inbound(@adapter, @conv, @user, "first") + assert await_text(@conv, "one") + first = Router.session_of(@adapter, @conv) + + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/new") + assert Router.session_of(@adapter, @conv) == nil + + Fake.scripts([script_deltas(1, "two ")]) + assert {:ok, :placed} = Router.inbound(@adapter, @conv, @user, "second") + assert await_text(@conv, "two") + assert Router.session_of(@adapter, @conv) != first + end + + test "a command that raises costs its own message and does not take the router down" do + :ok = pair!() + router = Process.whereis(Router) + + # `/attach` with something that is not an id: the repo would raise on the cast. + assert {:ok, :command} = Router.inbound(@adapter, @conv, @user, "/attach not-an-id") + assert Console.text(@conv) |> List.last() =~ "No session starts with not-an-id" + assert Process.whereis(Router) == router, "the router was restarted" + + # And the conversation still works afterwards. + Fake.scripts([script_deltas(1, "alive ")]) + assert {:ok, :placed} = Router.inbound(@adapter, @conv, @user, "still there?") + assert await_text(@conv, "alive") + end + + test "a revoked identity is turned away with a reason" do + :ok = pair!() + {:ok, identity} = Identities.revoke(Identities.get("console", @user)) + assert identity.state == "revoked" + + assert {:error, :revoked} = Router.inbound(@adapter, @conv, @user, "let me in") + assert Console.text(@conv) |> List.last() =~ "not allowed" + assert Fake.calls() == 0 + end +end diff --git a/test/trinity/gateways/stress_test.exs b/test/trinity/gateways/stress_test.exs new file mode 100644 index 0000000..c150c61 --- /dev/null +++ b/test/trinity/gateways/stress_test.exs @@ -0,0 +1,87 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Gateways.StressTest do + @moduledoc """ + Slice 070 AC7: fifty console conversations, each a paired identity of its own, all talking at + once under the fake provider. Every one gets its reply, every one has a session of its own, and + `PRAGMA integrity_check` is `ok` afterwards (on the Postgres leg the pragma does not apply and + the rest of the test is the whole of it). The router is one process, so this is also what says + it does not serialise into uselessness or leak a conversation's stream into another's. + """ + use Trinity.SessionCase + + alias Trinity.Gateways.{Console, Identities, Router} + alias Trinity.LLM.Providers.Fake + + @conversations 50 + + @tag timeout: 300_000 + test "AC7: fifty conversations complete, each in its own session, and the database is intact" do + start_supervised!(Console) + start_supervised!(Router) + Application.put_env(:trinity, :gateways, rate_limit: [capacity: 5, per_minute: 600]) + on_exit(fn -> Application.delete_env(:trinity, :gateways) end) + Fake.scripts([script_deltas(2, "reply ")]) + + identities = + for n <- 1..@conversations do + user = "u-#{n}" + {:ok, identity, :pending} = Identities.admit("console", user) + {:ok, _} = Identities.pair("console", user, identity.code) + {"c-#{n}", user} + end + + results = + identities + |> Task.async_stream( + fn {conversation, user} -> + Router.inbound(Console, conversation, user, "hello from #{conversation}") + end, + max_concurrency: @conversations, + timeout: 60_000 + ) + |> Enum.map(fn {:ok, result} -> result end) + + assert Enum.all?(results, &(&1 == {:ok, :placed})), "not every message was placed" + + # Every conversation is answered, and answered in its own channel. + for {conversation, _user} <- identities do + text = await_text(conversation) + assert text =~ "reply reply" + refute text =~ "hello from", "a conversation was shown another's message" + end + + # One session per conversation, each with the right origin, and nothing shared. + session_ids = + for {conversation, _} <- identities, do: Router.session_of(Console, conversation) + + assert length(Enum.uniq(session_ids)) == @conversations + assert Enum.all?(session_ids, &is_binary/1) + + for id <- session_ids do + session = Sessions.get_session(id) + assert session.origin == "console" + end + + # The repo's own adapter, as slice 010's stress test asks it: a configuration key can be set + # to something the repo is not. + if Trinity.Repo.__adapter__() == Ecto.Adapters.SQLite3 do + assert %{rows: [["ok"]]} = Trinity.Repo.query!("PRAGMA integrity_check") + end + end + + defp await_text(conversation, timeout \\ 60_000) do + deadline = System.monotonic_time(:millisecond) + timeout + do_await(conversation, deadline) + end + + defp do_await(conversation, deadline) do + text = Console.text(conversation) |> Enum.join("\n") + + cond do + text =~ "reply" -> text + System.monotonic_time(:millisecond) > deadline -> flunk("#{conversation} never answered") + true -> Process.sleep(50) && do_await(conversation, deadline) + end + end +end diff --git a/test/trinity_web/live/gateways_live_test.exs b/test/trinity_web/live/gateways_live_test.exs new file mode 100644 index 0000000..5fa070c --- /dev/null +++ b/test/trinity_web/live/gateways_live_test.exs @@ -0,0 +1,63 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule TrinityWeb.GatewaysLiveTest do + @moduledoc """ + Slice 070, AC9's automatic half: `/gateways` lists the configured channels with the tier each + may approve, shows a waiting identity's pairing code, and allows or revokes one. The code is + shown on this page and nowhere else, which is what makes reading it the proof of pairing. + """ + use TrinityWeb.ConnCase, async: false + + import Phoenix.LiveViewTest + + alias Trinity.Gateways.{Console, Identities} + + setup do + start_supervised!(Console) + Application.put_env(:trinity, :gateways, adapters: [Console]) + on_exit(fn -> Application.delete_env(:trinity, :gateways) end) + :ok + end + + test "the page lists the channel, whether it runs, and the tier it may approve", %{conn: conn} do + {:ok, _view, html} = live(conn, ~p"/gateways") + assert html =~ "console" + assert html =~ "running" + assert html =~ "approves up to" + assert html =~ "write" + assert html =~ "permissions page" + end + + test "a waiting identity's code is shown here, and allow pairs it without a code", %{conn: conn} do + {:ok, identity, :pending} = Identities.admit("console", "u-1", display_name: "Someone") + + {:ok, view, html} = live(conn, ~p"/gateways") + assert html =~ identity.code + assert html =~ "console:u-1" + assert html =~ "Someone" + + view |> element("#identity-#{identity.id} button", "allow") |> render_click() + assert Identities.get("console", "u-1").state == "paired" + refute render(view) =~ identity.code + end + + test "revoking keeps the row and shows it as revoked", %{conn: conn} do + {:ok, identity, :pending} = Identities.admit("console", "u-2") + + {:ok, view, _html} = live(conn, ~p"/gateways") + view |> element("#identity-#{identity.id} button", "revoke") |> render_click() + + assert Identities.get("console", "u-2").state == "revoked" + html = render(view) + assert html =~ "u-2" + assert html =~ "revoked" + end + + test "with no identity and no adapter configured the page says so", %{conn: conn} do + Application.put_env(:trinity, :gateways, adapters: []) + {:ok, _view, html} = live(conn, ~p"/gateways") + assert html =~ "No gateway is configured" + assert html =~ "Nobody is waiting" + assert html =~ "No identity has written yet" + end +end