aion-server 0.29.0

Aion workflow server library: HTTP, gRPC, WebSocket, and worker endpoints. Run it with the `aion` binary from the aion-cli crate.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
//! What the server says about the assistant it carries.
//!
//! The description is DERIVED on every read: the identity and contract come
//! from the embedded document's compiled form, and residency comes from the
//! engine catalog at request time. Nothing is remembered from boot, so a
//! description cannot go stale behind an operator's `aion deploy` or
//! `aion route` between restarts.

use aion::Engine;
use aion_integration_acp::catalogue;
use serde::Serialize;
use serde_json::Value;

use super::document::{
    CONTINUE_END_FIELD, CONTINUE_MESSAGE_FIELD, CONTINUE_SIGNAL, EMBEDDED_ASSISTANT_FILENAME,
    EmbeddedAssistant, OBJECTIVE_INPUT, REPO_PATH_INPUT, STATUS_QUERY,
};
use super::sessions::{AssistantSessionError, AssistantSessions};
use crate::namespace::CallerIdentity;
use crate::namespace::grants::GRANT_WORDS;

/// Where the embedded document stands in the engine catalog right now.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case", tag = "state")]
pub enum AssistantResidency {
    /// The embedded version is loaded and holds the route: starting the
    /// assistant runs this document.
    Routed,
    /// The embedded version is loaded but another version holds the route (or
    /// nothing does). Starting the assistant does not run this document.
    LoadedNotRouted {
        /// The routed hash, when some resident version holds the route.
        routed_hash: Option<String>,
    },
    /// The embedded version is not in the catalog at all. `routed_hash` names
    /// whatever version of the type does hold the route; `None` means the type
    /// is unstartable on this server.
    NotLoaded {
        /// The routed hash, when some other version holds the route.
        routed_hash: Option<String>,
    },
    /// The catalog could not be read, so residency is unknown — never guessed.
    Unknown {
        /// Why the catalog could not be read.
        reason: String,
    },
}

/// The names the operator surfaces bind to, verified against the document at
/// load time. Published so a client drives the session by reading rather than
/// by restating them.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantSessionContract {
    /// Start input carrying the operator's opening ask.
    pub objective_input: &'static str,
    /// Start input carrying the repository path (empty = scratch mode).
    pub repo_path_input: &'static str,
    /// The signal a parked session listens on.
    pub continue_signal: &'static str,
    /// Continuation field carrying the next prompt.
    pub message_field: &'static str,
    /// Continuation field that ends the session.
    pub end_field: &'static str,
    /// The read-only status query.
    pub status_query: &'static str,
}

impl AssistantSessionContract {
    /// The contract this binary was built with.
    #[must_use]
    pub const fn current() -> Self {
        Self {
            objective_input: OBJECTIVE_INPUT,
            repo_path_input: REPO_PATH_INPUT,
            continue_signal: CONTINUE_SIGNAL,
            message_field: CONTINUE_MESSAGE_FIELD,
            end_field: CONTINUE_END_FIELD,
            status_query: STATUS_QUERY,
        }
    }
}

/// One declared signal and its payload schema.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantSignal {
    /// Signal name.
    pub name: String,
    /// Payload schema, draft 2020-12, derived from the document.
    pub input_schema: Value,
}

/// One harness this server can open an assistant session on.
///
/// The list IS the catalogue this build ships
/// ([`aion_integration_acp::catalogue`]) — an operator never types a command —
/// and each entry carries the two facts a picker needs beside the name: whether
/// this machine can actually run it, and what to install when it cannot.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantHarnessDescriptor {
    /// The catalogue id, which a create request selects by.
    pub name: String,
    /// The adapter that runs it.
    pub kind: String,
    /// The declared login account names, in declaration order. Empty when the
    /// harness declares none, which is a complete answer: a session on it then
    /// names no account.
    pub accounts: Vec<String>,
    /// Whether [`Self::launch`]'s program resolves on THIS server's `PATH`,
    /// measured while this description was being built.
    ///
    /// Never cached: an operator who installs Node.js and reloads the console
    /// must see the entry come alive without restarting the server, and a server
    /// that remembered a boot-time reading would be reporting a machine as it
    /// was rather than as it is.
    pub available: bool,
    /// The catalogue's own sentence naming what to install, or `None` when the
    /// harness is available. Present exactly when [`Self::available`] is false,
    /// so a client renders a hint or nothing and never both.
    pub install_hint: Option<String>,
    /// The exact line this server would run for it. DISPLAY only: it is shown
    /// so an operator can see what a session starts, and it is not something any
    /// request may set.
    pub launch: String,
}

/// The tool wiring every session on this server is handed.
///
/// NAMES only, never a command line or a URL: what tools an agent is given is
/// something an operator must be able to see from the console, and how they are
/// reached is a spawn detail that would leak a local path or an internal
/// endpoint to every reader of this description.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantToolsDescriptor {
    /// Whether this server's own general MCP endpoint is handed to the harness
    /// — the aion tools an agent drives workflows with.
    pub aion: bool,
    /// The assistant's OWN tool server: the second MCP endpoint every session's
    /// agent is handed, separate from the general one.
    pub assistant: AssistantOwnToolsDescriptor,
}

/// The assistant-only MCP server handed to every session's agent.
///
/// Published so an operator can see, without reading the source, that a session
/// hands its agent a tool for reading the console screen — what it is called,
/// where it is served, and what credential it takes.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantOwnToolsDescriptor {
    /// The name the agent shows for it.
    pub server: String,
    /// The route it is served on.
    pub route: String,
    /// Every tool in its catalogue. Exactly one, and it is not in the general
    /// catalogue — a general caller asking for it is refused by name.
    pub tools: Vec<String>,
    /// Whether it is actually handed over on this server. `false` when no
    /// dialable address can be stated (a configured port of zero).
    pub handed_over: bool,
    /// Why it is not handed over, or `None` when it is.
    pub unavailable_reason: Option<String>,
    /// The credential it accepts.
    pub token: AssistantSessionTokenDescriptor,
}

/// The credential the assistant tool route accepts.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantSessionTokenDescriptor {
    /// The wire word for the kind, for a client that branches on it.
    pub kind: String,
    /// Who mints it: this server, never a caller.
    pub minted_by: String,
    /// What it authorizes: one session, never a namespace or a person.
    pub scope: String,
    /// The one sentence describing it, from
    /// [`crate::assistant::mcp::SESSION_TOKEN_DESCRIPTION`] — the SAME sentence
    /// `docs/operations/API.md` quotes, so there is one description of one
    /// credential rather than two that could drift.
    pub description: String,
}

impl AssistantSessionTokenDescriptor {
    /// The description of the credential this binary mints.
    #[must_use]
    pub fn current() -> Self {
        Self {
            kind: crate::assistant::mcp::SESSION_TOKEN_KIND.to_owned(),
            minted_by: TOKEN_MINTED_BY.to_owned(),
            scope: TOKEN_SCOPE.to_owned(),
            description: crate::assistant::mcp::SESSION_TOKEN_DESCRIPTION.to_owned(),
        }
    }
}

/// Who mints the session bearer. The server, always: a caller cannot present
/// one it made, because verification is against a digest only the server wrote.
const TOKEN_MINTED_BY: &str = "server";
/// What the session bearer authorizes.
const TOKEN_SCOPE: &str = "session";
/// Why the assistant tool server is not handed over when it is not.
const NO_DIALABLE_ADDRESS: &str = "this server cannot state an address an agent could dial back on (`server.listen_address` \
     names port 0, whose real port is only known after bind), so no MCP server of ours is handed \
     to a session's agent and it cannot read what is on the operator's screen";

/// One grant word this deployment defines, and whether the caller reading this
/// description holds it.
///
/// Built by walking [`GRANT_WORDS`], never a hand-written list: a word that
/// existed in the grammar and not here would be grantable and undiscoverable.
/// The `description` is the grammar row's own sentence, so the console states
/// what a word authorises rather than inventing a meaning for it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantGrantDescriptor {
    /// The stable word an operator and an audit line spell.
    pub name: String,
    /// Whether this caller holds it.
    pub held: bool,
    /// One sentence naming what the word authorises, from the grammar row.
    pub description: String,
}

/// The adapter kind every resolved harness runs.
///
/// [`crate::config::ResolvedAssistantHarness`] carries no kind because there is
/// nothing per-harness left to remember: `config/assistant_resolve.rs` refuses
/// every value but this one at load, so a RESOLVED harness is an ACP harness by
/// construction. It is published anyway — a client must not have to know that
/// rule to read this description — and pinned against the resolver itself by
/// `the_published_harness_kind_is_the_one_resolution_accepts` in
/// `api/http/assistant_sessions_tests.rs`, so the two cannot drift.
const HARNESS_KIND: &str = "acp";

/// The name the agent shows for the assistant's own tool server.
///
/// Pinned against `launch.rs`'s own constant by
/// `the_published_assistant_server_name_is_the_one_a_spawn_hands_over`, so the
/// description and the spawn cannot name two different servers.
const ASSISTANT_TOOL_SERVER_NAME: &str = "assistant";

/// The served description of the built-in assistant.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AssistantDescriptor {
    /// The workflow type an operator starts.
    pub workflow_type: String,
    /// The private task queue a worker must serve for a session to run.
    ///
    /// Published because an operator with no checkout has no other way to
    /// learn it, and because getting it wrong is silent: a worker on the wrong
    /// queue connects, registers, and never receives a dispatch. Derived from
    /// the document's own identity and never `default` — see
    /// [`super::document::private_task_queue`].
    pub task_queue: String,
    /// The embedded document's content hash — its version identity.
    pub content_hash: String,
    /// The document's filename inside the binary.
    pub document_filename: &'static str,
    /// Length of the embedded document in bytes, so a client can tell a
    /// truncated fetch of `/assistant/document` from a complete one.
    pub document_bytes: usize,
    /// Start input schema, draft 2020-12, derived from the document.
    pub input_schema: Value,
    /// Every declared signal with its payload schema.
    pub signals: Vec<AssistantSignal>,
    /// Every declared query name, in document order.
    pub queries: Vec<String>,
    /// The verified session contract.
    pub session: AssistantSessionContract,
    /// Where the embedded version stands in the catalog right now.
    pub residency: AssistantResidency,
    /// Every harness this server can open an assistant session on — the
    /// build's catalogue, in catalogue order, each with its availability on this
    /// machine.
    pub harnesses: Vec<AssistantHarnessDescriptor>,
    /// The harness THIS caller last opened a session on, or `None` before they
    /// have opened one.
    ///
    /// A memory, not a policy: it is written by `createSession` from what the
    /// operator actually did, and it is caller-scoped, so one operator's habit
    /// never preselects another's console. `None` is a complete answer — a
    /// client preselects the first available entry rather than the server
    /// inventing a choice nobody made.
    pub default_harness: Option<String>,
    /// The tool wiring every session on this server is handed.
    pub tools: AssistantToolsDescriptor,
    /// Whether this server can open an assistant session at all.
    pub sessions_enabled: bool,
    /// Why it cannot, in words an operator can act on, or `None` when it can.
    ///
    /// Reserved for a refusal the PRODUCT can name — today, a durable store this
    /// server could not read at start-up. It is never "not configured": a stock
    /// server with no `[assistant]` section serves the assistant, and a harness
    /// this machine cannot run is said per entry, with its install hint, rather
    /// than as one sentence about the whole surface.
    pub sessions_disabled_reason: Option<String>,
    /// Every grant word this deployment defines, and whether the caller reading
    /// this description holds it.
    pub grants: Vec<AssistantGrantDescriptor>,
}

/// Describes `embedded` against `engine`'s current catalog, `sessions`'
/// configuration and availability, and what `caller` is authorized to do.
///
/// Everything is DERIVED at request time and nothing is remembered from boot —
/// including the grant rows, which are read off this request's own resolved
/// identity, so one description can never report another caller's
/// authorization.
///
/// # Errors
///
/// Whatever the store reports while reading this caller's last harness pick.
/// The pick is a stored fact about the caller, so it is read rather than
/// remembered, and a store that cannot answer is reported rather than being
/// rendered as "no pick" — which is a different, and wrong, thing to tell a
/// console.
pub async fn describe(
    embedded: &EmbeddedAssistant,
    engine: &Engine,
    sessions: &AssistantSessions,
    caller: &CallerIdentity,
) -> Result<AssistantDescriptor, AssistantSessionError> {
    let config = sessions.config();
    let availability = sessions.availability();
    let default_harness = sessions.last_harness_pick(caller.subject()).await?;
    Ok(AssistantDescriptor {
        workflow_type: embedded.workflow_type().to_owned(),
        task_queue: embedded.task_queue().to_owned(),
        content_hash: embedded.content_hash().to_string(),
        document_filename: EMBEDDED_ASSISTANT_FILENAME,
        document_bytes: embedded.source().len(),
        input_schema: embedded.input_schema().clone(),
        signals: embedded
            .signals()
            .iter()
            .map(|signal| AssistantSignal {
                name: signal.name.clone(),
                input_schema: signal.input_schema.clone(),
            })
            .collect(),
        queries: embedded.queries().to_vec(),
        session: AssistantSessionContract::current(),
        residency: residency(embedded, engine),
        harnesses: catalogue::CATALOGUE
            .iter()
            .map(|entry| AssistantHarnessDescriptor {
                name: entry.id.to_owned(),
                kind: HARNESS_KIND.to_owned(),
                accounts: config.account_names(entry.id),
                // MEASURED here, on every read. The hint is carried exactly when
                // it is needed, so a client cannot render "install Node.js"
                // beside a harness that is already running.
                available: entry.available(),
                install_hint: (!entry.available()).then(|| entry.install_hint.to_owned()),
                launch: entry.launch(),
            })
            .collect(),
        default_harness,
        tools: AssistantToolsDescriptor {
            aion: sessions.hands_over_general_mcp(),
            assistant: AssistantOwnToolsDescriptor {
                server: ASSISTANT_TOOL_SERVER_NAME.to_owned(),
                route: crate::assistant::mcp::ASSISTANT_MCP_PATH.to_owned(),
                tools: crate::assistant::mcp::SESSION_TOOL_NAMES
                    .iter()
                    .map(|name| (*name).to_owned())
                    .collect(),
                handed_over: sessions.hands_over_assistant_tools(),
                unavailable_reason: (!sessions.hands_over_assistant_tools())
                    .then(|| NO_DIALABLE_ADDRESS.to_owned()),
                token: AssistantSessionTokenDescriptor::current(),
            },
        },
        sessions_enabled: availability.is_available(),
        sessions_disabled_reason: availability.reason().map(ToOwned::to_owned),
        grants: GRANT_WORDS
            .iter()
            .map(|grant| AssistantGrantDescriptor {
                name: grant.word().to_owned(),
                held: grant.granted_for(caller),
                description: grant.description().to_owned(),
            })
            .collect(),
    })
}

/// Reads the catalog and classifies the embedded version's standing.
fn residency(embedded: &EmbeddedAssistant, engine: &Engine) -> AssistantResidency {
    let versions = match engine.list_workflow_versions() {
        Ok(versions) => versions,
        Err(error) => {
            return AssistantResidency::Unknown {
                reason: format!("the engine catalog could not be read: {error}"),
            };
        }
    };
    let embedded_hash = embedded.content_hash().to_string();
    let resident: Vec<_> = versions
        .into_iter()
        .filter(|version| version.workflow_type == embedded.workflow_type())
        .collect();
    let loaded = resident
        .iter()
        .any(|version| version.content_hash.to_string() == embedded_hash);
    let routed_hash = resident
        .iter()
        .find(|version| version.route_active)
        .map(|version| version.content_hash.to_string());
    if !loaded {
        return AssistantResidency::NotLoaded { routed_hash };
    }
    if routed_hash.as_deref() == Some(embedded_hash.as_str()) {
        AssistantResidency::Routed
    } else {
        AssistantResidency::LoadedNotRouted { routed_hash }
    }
}