Skip to content

Commit 3588c70

Browse files
committed
feat(permissions): wire actor permission requests and warn on missing handler
- Broadcast `permissionRequest` events per session and add a `respondPermission` action in the actor plugin so the RivetKit client-side approval flow works. - Warn once per session (host-visible, via onAgentStderr) when a tool-permission request arrives with no onPermissionRequest handler registered, instead of a silent deny. - Update the Approvals and crash-course docs and examples to the client-side permissionRequest/respondPermission API; drop the non-functional server-hook auto-approve pattern.
1 parent cd4c53a commit 3588c70

12 files changed

Lines changed: 516 additions & 104 deletions

File tree

‎crates/agentos-actor-plugin/src/actions/mod.rs‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,8 @@ pub struct Vars {
3939
pub live_sessions: HashMap<String, String>,
4040
/// `live_session_id -> capture pump task`.
4141
pub capture_tasks: HashMap<String, JoinHandle<()>>,
42+
/// `live_session_id -> permission-request pump task`.
43+
pub permission_tasks: HashMap<String, JoinHandle<()>>,
4244
}
4345

4446
impl Vars {
@@ -51,12 +53,15 @@ impl Vars {
5153
.unwrap_or(external_session_id)
5254
}
5355

54-
/// Abort and clear all in-flight capture tasks. Called on VM teardown
55-
/// (sleep / destroy / run-loop exit).
56+
/// Abort and clear all in-flight pump tasks (event capture + permission
57+
/// requests). Called on VM teardown (sleep / destroy / run-loop exit).
5658
pub fn clear(&mut self) {
5759
for (_, task) in self.capture_tasks.drain() {
5860
task.abort();
5961
}
62+
for (_, task) in self.permission_tasks.drain() {
63+
task.abort();
64+
}
6065
self.live_sessions.clear();
6166
}
6267
}
@@ -336,6 +341,17 @@ pub(crate) async fn dispatch(
336341
},
337342
Err(error) => reply_err(host, token, error),
338343
},
344+
"respondPermission" => match decode_as::<(String, String, String)>(args) {
345+
Ok((session_id, permission_id, reply)) => {
346+
match session::respond_permission(vm, vars, &session_id, &permission_id, &reply)
347+
.await
348+
{
349+
Ok(()) => reply_ok(host, token, &()),
350+
Err(error) => reply_err(host, token, error),
351+
}
352+
}
353+
Err(error) => reply_err(host, token, error),
354+
},
339355
"createSignedPreviewUrl" => match decode_as::<(u16, u64)>(args) {
340356
Ok((port, ttl_seconds)) => match preview::create(host, port, ttl_seconds).await {
341357
Ok(dto) => reply_ok(host, token, &dto),

‎crates/agentos-actor-plugin/src/actions/session.rs‎

Lines changed: 169 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ use std::collections::BTreeMap;
1111
use std::time::{SystemTime, UNIX_EPOCH};
1212

1313
use crate::host_ctx::HostCtx;
14-
use agentos_client::{AgentOs, CreateSessionOptions};
14+
use agentos_client::{AgentOs, CreateSessionOptions, PermissionReply};
1515
use anyhow::{anyhow, Result};
1616
use futures::StreamExt;
1717
use serde::{Deserialize, Serialize};
@@ -135,6 +135,111 @@ fn spawn_event_capture(
135135
.insert(live_session_id.to_owned(), handle);
136136
}
137137

138+
/// Build the `permissionRequest` broadcast body for one request.
139+
///
140+
/// The RivetKit event wire is CBOR and the body is the array of handler
141+
/// ARGUMENTS the client spreads into the listener (`handler(...body)`). The
142+
/// documented listener is `(data) => …`, so the single argument is the TS
143+
/// `PermissionRequestPayload` — `{ sessionId, request: { permissionId,
144+
/// description?, params } }` — and the body is `[ <that object> ]`. The
145+
/// `sessionId` is the client-facing external id (== live for native sessions).
146+
fn permission_event_body(
147+
external_session_id: &str,
148+
permission_id: &str,
149+
description: Option<&str>,
150+
params: &JsonValue,
151+
) -> JsonValue {
152+
json!([{
153+
"sessionId": external_session_id,
154+
"request": {
155+
"permissionId": permission_id,
156+
"description": description,
157+
"params": params,
158+
},
159+
}])
160+
}
161+
162+
/// Map the wire reply string to a [`PermissionReply`] (`"once"` / `"always"` /
163+
/// `"reject"`), matching the TS `PermissionReply` union.
164+
fn parse_permission_reply(reply: &str) -> Result<PermissionReply> {
165+
match reply {
166+
"once" => Ok(PermissionReply::Once),
167+
"always" => Ok(PermissionReply::Always),
168+
"reject" => Ok(PermissionReply::Reject),
169+
other => Err(anyhow!(
170+
"invalid permission reply {other:?} (expected \"once\" | \"always\" | \"reject\")"
171+
)),
172+
}
173+
}
174+
175+
/// Subscribe to the session's permission-request stream and spawn a task that
176+
/// broadcasts each request to connected clients as a `permissionRequest` event
177+
/// (`conn.on("permissionRequest", …)`).
178+
///
179+
/// Mirrors [`spawn_event_capture`]. A subscriber MUST exist before the guest
180+
/// agent raises a permission request, otherwise the client auto-rejects it
181+
/// (`deliver_sidecar_permission_request` checks `receiver_count() == 0`) — so
182+
/// this is started at session-create time. Clients answer via the
183+
/// `respondPermission` action (→ [`respond_permission`]), which resolves the
184+
/// pending reply slot; this pump only fans the request out, so dropping the
185+
/// broadcast item's responder clone here is harmless.
186+
fn spawn_permission_pump(
187+
ctx: &HostCtx,
188+
vm: &AgentOs,
189+
vars: &mut Vars,
190+
external_session_id: &str,
191+
live_session_id: &str,
192+
) {
193+
let (mut stream, subscription) = match vm.on_permission_request(live_session_id) {
194+
Ok(sub) => sub,
195+
Err(error) => {
196+
tracing::warn!(?error, live_session_id, "on_permission_request subscribe failed");
197+
return;
198+
}
199+
};
200+
if let Some(old) = vars.permission_tasks.remove(live_session_id) {
201+
old.abort();
202+
}
203+
let ctx = ctx.clone();
204+
let external = external_session_id.to_owned();
205+
let handle = tokio::spawn(async move {
206+
// Keep the RAII guard alive for the pump's lifetime; dropping the stream
207+
// (on abort / channel close) is the unsubscribe.
208+
let _subscription = subscription;
209+
while let Some(request) = stream.next().await {
210+
let body = permission_event_body(
211+
&external,
212+
&request.permission_id,
213+
request.description.as_deref(),
214+
&request.params,
215+
);
216+
let mut cbor = Vec::new();
217+
if ciborium::into_writer(&body, &mut cbor).is_ok() {
218+
let _ = ctx.broadcast(b"permissionRequest".to_vec(), cbor);
219+
}
220+
}
221+
});
222+
vars.permission_tasks
223+
.insert(live_session_id.to_owned(), handle);
224+
}
225+
226+
/// Answer a permission request raised by the session's guest agent
227+
/// (`respondPermission`). Resolves the pending reply slot through the client's
228+
/// `respond_permission`, keyed by the live session id.
229+
pub async fn respond_permission(
230+
vm: &AgentOs,
231+
vars: &Vars,
232+
session_id: &str,
233+
permission_id: &str,
234+
reply: &str,
235+
) -> Result<()> {
236+
let reply = parse_permission_reply(reply)?;
237+
let live_session_id = vars.live_id(session_id).to_owned();
238+
vm.respond_permission(&live_session_id, permission_id, reply)
239+
.await?;
240+
Ok(())
241+
}
242+
138243
pub async fn create_session(
139244
ctx: &HostCtx,
140245
vm: &AgentOs,
@@ -176,8 +281,11 @@ pub async fn create_session(
176281
)
177282
.await?;
178283
// At create time `external == live`; capture every `session/update` for this
179-
// session under the external id (spec §3/§5).
284+
// session under the external id (spec §3/§5), and start fanning the guest's
285+
// permission requests out to connected clients. The permission pump must be
286+
// subscribed before the agent runs, or requests would auto-reject.
180287
spawn_event_capture(ctx, vm, vars, &session_id, &session_id);
288+
spawn_permission_pump(ctx, vm, vars, &session_id, &session_id);
181289
Ok(SessionIdDto { session_id })
182290
}
183291

@@ -227,11 +335,14 @@ pub async fn close_session(
227335
vars: &mut Vars,
228336
session_id: &str,
229337
) -> Result<()> {
230-
// Stop event capture + drop the remap for this external session.
338+
// Stop event capture + the permission pump + drop the remap for this session.
231339
let live_session_id = vars.live_id(session_id).to_owned();
232340
if let Some(task) = vars.capture_tasks.remove(&live_session_id) {
233341
task.abort();
234342
}
343+
if let Some(task) = vars.permission_tasks.remove(&live_session_id) {
344+
task.abort();
345+
}
235346
vars.live_sessions.remove(session_id);
236347
vm.close_session(&live_session_id).map_err(|e| anyhow!(e))?;
237348
// Drop persisted metadata + events (explicit, since SQLite FK cascade is
@@ -397,3 +508,58 @@ pub async fn resume_session(
397508
AcpResumeSessionRequest contract"
398509
))
399510
}
511+
512+
#[cfg(test)]
513+
mod tests {
514+
use super::*;
515+
use agentos_client::PermissionReply;
516+
517+
#[test]
518+
fn permission_event_body_matches_ts_payload_shape() {
519+
// The TS client listener is `(data) => …` where data is
520+
// `PermissionRequestPayload { sessionId, request: { permissionId,
521+
// description?, params } }`, delivered as the single broadcast arg.
522+
let params = json!({
523+
"toolCall": { "title": "Bash", "kind": "execute" },
524+
"options": [{ "optionId": "allow_once" }],
525+
});
526+
let body = permission_event_body("sess-1", "perm-7", Some("run a command"), &params);
527+
528+
// Body is the args array spread into the listener: exactly one argument.
529+
let args = body.as_array().expect("body is an array");
530+
assert_eq!(args.len(), 1, "exactly one handler argument");
531+
let data = &args[0];
532+
533+
assert_eq!(data["sessionId"], json!("sess-1"));
534+
assert_eq!(data["request"]["permissionId"], json!("perm-7"));
535+
assert_eq!(data["request"]["description"], json!("run a command"));
536+
// params are forwarded verbatim so the client can inspect the tool/paths.
537+
assert_eq!(data["request"]["params"], params);
538+
}
539+
540+
#[test]
541+
fn permission_event_body_serializes_absent_description_as_null() {
542+
let body = permission_event_body("sess-1", "perm-1", None, &json!({}));
543+
assert_eq!(body[0]["request"]["description"], JsonValue::Null);
544+
}
545+
546+
#[test]
547+
fn parse_permission_reply_maps_each_wire_value() {
548+
assert_eq!(parse_permission_reply("once").unwrap(), PermissionReply::Once);
549+
assert_eq!(
550+
parse_permission_reply("always").unwrap(),
551+
PermissionReply::Always
552+
);
553+
assert_eq!(
554+
parse_permission_reply("reject").unwrap(),
555+
PermissionReply::Reject
556+
);
557+
}
558+
559+
#[test]
560+
fn parse_permission_reply_rejects_unknown_value() {
561+
let err = parse_permission_reply("maybe").unwrap_err().to_string();
562+
assert!(err.contains("invalid permission reply"), "got: {err}");
563+
assert!(err.contains("maybe"), "names the bad value: {err}");
564+
}
565+
}

‎examples/docs/approvals/auto-approve-client.ts‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,13 @@ import type { registry } from "./server";
44
const client = createClient<typeof registry>({ endpoint: "http://localhost:6420" });
55
const agent = client.vm.getOrCreate("my-agent");
66

7-
// No need to handle permissions on the client. The server hook handles them.
7+
// Auto-approve every request as it arrives. `"always"` also approves future
8+
// requests of the same type, so a multi-step agent run is not interrupted.
9+
const conn = agent.connect();
10+
conn.on("permissionRequest", async (data) => {
11+
await agent.respondPermission(data.sessionId, data.request.permissionId, "always");
12+
});
13+
814
const session = await agent.createSession("claude", {
915
env: { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY! },
1016
});

‎examples/docs/approvals/auto-approve.ts‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,6 @@ import pi from "./software/pi";
33

44
const vm = agentOS({
55
software: [pi],
6-
// The onPermissionRequest hook runs server-side for every request before it
7-
// is forwarded to clients. Use it to inspect requests in fully automated
8-
// pipelines without a client round-trip.
9-
onPermissionRequest: async (sessionId, request) => {
10-
console.log("auto-approving", sessionId, request.permissionId);
11-
},
126
});
137

148
export const registry = setup({ use: { vm } });

‎examples/docs/approvals/selective-client.ts‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,17 @@ import type { registry } from "./server";
44
const client = createClient<typeof registry>({ endpoint: "http://localhost:6420" });
55
const agent = client.vm.getOrCreate("my-agent");
66

7-
// Permission requests forwarded by the server reach the client here. The
8-
// payload is inferred from the actor's event schema, so no cast is needed.
97
const conn = agent.connect();
108
conn.on("permissionRequest", async (data) => {
9+
// Inspect the request and decide per-request. `request.description` /
10+
// `request.params` carry the raw ACP details (the requested tool, paths, etc.).
11+
const description = data.request.description?.toLowerCase() ?? "";
12+
if (description.includes("read")) {
13+
// Auto-approve reads.
14+
await agent.respondPermission(data.sessionId, data.request.permissionId, "always");
15+
return;
16+
}
17+
// Forward everything else to a human.
1118
const approved = confirm(`Allow: ${JSON.stringify(data.request)}?`);
1219
await agent.respondPermission(
1320
data.sessionId,

‎examples/docs/approvals/selective.ts‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,15 +3,6 @@ import pi from "./software/pi";
33

44
const vm = agentOS({
55
software: [pi],
6-
onPermissionRequest: async (sessionId, request) => {
7-
// `request.description` and `request.params` carry the raw ACP permission
8-
// details (the requested tool, paths, etc.). Inspect them to decide which
9-
// requests to handle server-side and which to forward to clients.
10-
const description = request.description ?? "";
11-
if (description.toLowerCase().includes("read")) {
12-
console.log("read request handled server-side", sessionId, request.permissionId);
13-
}
14-
},
156
});
167

178
export const registry = setup({ use: { vm } });

‎examples/docs/crash-course/permissions-client.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,12 @@ import type { registry } from "./server";
44
const client = createClient<typeof registry>({ endpoint: "http://localhost:6420" });
55
const agent = client.vm.getOrCreate("my-agent");
66

7-
// Or handle permissions client-side for human-in-the-loop
7+
// Subscribe and reply to permission requests. Permissions are fail-closed, so
8+
// the agent waits until you reply.
89
const conn = agent.connect();
910
conn.on("permissionRequest", async (data) => {
1011
console.log("Permission requested:", data.request);
11-
// "once" | "always" | "reject"
12+
// "once" | "always" | "reject". Reply "always" to auto-approve trusted
13+
// workloads, or prompt a human for human-in-the-loop.
1214
await agent.respondPermission(data.sessionId, data.request.permissionId, "once");
1315
});

‎examples/docs/crash-course/permissions-server.ts‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,8 @@
11
import { agentOS, setup } from "@rivet-dev/agentos";
22
import pi from "./software/pi";
33

4-
// Auto-approve all permissions server-side
54
const vm = agentOS({
65
software: [pi],
7-
onPermissionRequest: async (sessionId, request) => {
8-
console.log("Auto-approving", sessionId, request.permissionId);
9-
},
106
});
117

128
export const registry = setup({ use: { vm } });

0 commit comments

Comments
 (0)