feat: connect restricted async chat to background Worker results
Exercise public Spawn, real bounded work, other-conversation handling, ordinary Notify and explicit publication to the originating message. Retain stable complete spawn input across acknowledgement loss and retries; verify operation and target revocation without a second result or grant registry.
This commit is contained in:
@@ -0,0 +1,692 @@
|
||||
//! Async-chat consumer integration (`cargo test -p yoi-workspace-server --lib
|
||||
//! chat_delegation_connected`). Public chat/Worker tools, signed Backend/Runtime,
|
||||
//! Controller, grants, SQLite, Session and Workdir effects are real. Only the
|
||||
//! external model is scripted/gated; the shared RAII harness owns all resources.
|
||||
use super::*;
|
||||
use agen::llm_client::{
|
||||
event::{Event, StopReason},
|
||||
types::Item,
|
||||
};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use worker_runtime::worker_backend::RuntimeWorkerFactory;
|
||||
use worker_runtime::working_directory::WorkingDirectoryMaterializer;
|
||||
|
||||
const CHILD: &str = "BOUNDED_CHAT_BACKGROUND_WORK";
|
||||
const REPORT: &str = "BACKGROUND_EVIDENCE_FOR_ORIGINAL_REQUEST";
|
||||
const PUBLIC_A: &str = "Evaluated background evidence for A";
|
||||
const PUBLIC_B: &str = "Independent answer for B";
|
||||
|
||||
fn operation(name: &str, collection: bool) -> ScopedPermission {
|
||||
ScopedPermission {
|
||||
operation: name.into(),
|
||||
resource: if collection {
|
||||
PermissionResource::Workspace {
|
||||
workspace_id: TEST_WORKSPACE_ID.into(),
|
||||
}
|
||||
} else {
|
||||
PermissionResource::WorkerOperation {
|
||||
workspace_id: TEST_WORKSPACE_ID.into(),
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
fn output(request: &agen::llm_client::Request, id: &str) -> Option<(bool, String)> {
|
||||
request.items.iter().find_map(|item| match item {
|
||||
Item::ToolResult {
|
||||
call_id,
|
||||
is_error,
|
||||
content,
|
||||
summary,
|
||||
..
|
||||
} if call_id == id => Some((
|
||||
*is_error,
|
||||
content.clone().unwrap_or_else(|| summary.clone()),
|
||||
)),
|
||||
_ => None,
|
||||
})
|
||||
}
|
||||
fn value(request: &agen::llm_client::Request, id: &str) -> Value {
|
||||
let (error, text) = output(request, id).unwrap_or_else(|| panic!("missing tool result {id}"));
|
||||
assert!(!error, "{id}: {text}");
|
||||
serde_json::from_str(&text).unwrap_or_else(|_| panic!("{id}: {text}"))
|
||||
}
|
||||
fn call(id: &str, tool: &str, input: Value) -> Vec<Event> {
|
||||
vec![
|
||||
Event::tool_use_start(0, id, tool),
|
||||
Event::tool_input_delta(0, input.to_string()),
|
||||
Event::tool_use_stop(0),
|
||||
]
|
||||
}
|
||||
fn done() -> Vec<Event> {
|
||||
vec![
|
||||
Event::text_block_start(0),
|
||||
Event::text_delta(0, "PRIVATE_CHAT_AND_WORKER_OUTPUT"),
|
||||
Event::text_block_stop(0, Some(StopReason::EndTurn)),
|
||||
]
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Context {
|
||||
a: String,
|
||||
b: String,
|
||||
workdir: Option<String>,
|
||||
permissions: Vec<ScopedPermission>,
|
||||
spawned: Option<RuntimeWorkerRef>,
|
||||
}
|
||||
#[derive(Clone)]
|
||||
struct DelegationModel {
|
||||
context: Arc<Mutex<Context>>,
|
||||
body_phase: Arc<AtomicUsize>,
|
||||
child_phase: Arc<AtomicUsize>,
|
||||
child_started: tokio::sync::mpsc::Sender<()>,
|
||||
release_work: Arc<tokio::sync::Semaphore>,
|
||||
completed: tokio::sync::mpsc::Sender<()>,
|
||||
private_report: tokio::sync::mpsc::Sender<()>,
|
||||
release_send: Arc<tokio::sync::Semaphore>,
|
||||
lose_spawn_ack: bool,
|
||||
}
|
||||
impl DelegationModel {
|
||||
fn spawn(&self, page: &Value) -> Value {
|
||||
let c = self.context.lock().unwrap();
|
||||
let message = page["conversation"]["last_user_message_id"]
|
||||
.as_str()
|
||||
.unwrap();
|
||||
let mut input = json!({"runtime_id":CONNECTED_RUNTIME,"profile":"project:chat-background",
|
||||
"operation_id":format!("chat:{}:{message}", c.a),
|
||||
"display_name":"Bounded background chat work",
|
||||
"initial_submit":[{"kind":"text","content":format!("{CHILD}; conversation={}; message={message}", c.a)}],
|
||||
"execution_grant":{"permissions":c.permissions,"redelegation":[]}});
|
||||
if let Some(id) = &c.workdir {
|
||||
input["working_directory_id"] = json!(id);
|
||||
}
|
||||
input
|
||||
}
|
||||
fn send(
|
||||
&self,
|
||||
request: &agen::llm_client::Request,
|
||||
read: &str,
|
||||
text: &str,
|
||||
operation: &str,
|
||||
work: bool,
|
||||
) -> Value {
|
||||
let page = value(request, read);
|
||||
let message = &page["conversation"]["last_user_message_id"];
|
||||
let mut input = json!({"conversation_id":page["conversation"]["id"],"operation_id":operation,
|
||||
"text":text,"reply_to":message,"expected_last_user_message_id":message});
|
||||
if work {
|
||||
let child = value(request, "spawn-retry");
|
||||
input["work_ref"] = json!(format!(
|
||||
"worker:{}/{}",
|
||||
child["runtime_id"].as_str().unwrap(),
|
||||
child["worker_id"].as_str().unwrap()
|
||||
));
|
||||
}
|
||||
input
|
||||
}
|
||||
}
|
||||
#[async_trait::async_trait]
|
||||
impl agen::llm_client::LlmClient for DelegationModel {
|
||||
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
|
||||
Box::new(self.clone())
|
||||
}
|
||||
async fn stream(
|
||||
&self,
|
||||
request: agen::llm_client::Request,
|
||||
) -> std::result::Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError>
|
||||
{
|
||||
let child = request.items.iter().any(|item| matches!(item, Item::Message { content, .. } if serde_json::to_string(content).unwrap().contains(CHILD)));
|
||||
let events = if child {
|
||||
let phase = self.child_phase.fetch_add(1, Ordering::SeqCst);
|
||||
let workdir = self.context.lock().unwrap().workdir.is_some();
|
||||
match phase {
|
||||
0 => {
|
||||
self.child_started.try_send(()).unwrap();
|
||||
tokio::time::timeout(
|
||||
std::time::Duration::from_secs(60),
|
||||
self.release_work.acquire(),
|
||||
)
|
||||
.await
|
||||
.expect("B must be answered before releasing background work")
|
||||
.unwrap()
|
||||
.forget();
|
||||
if workdir {
|
||||
call("work-read", "Read", json!({"file_path":"README.md"}))
|
||||
} else {
|
||||
call("work-query", "QueryObjective", json!({"limit":1}))
|
||||
}
|
||||
}
|
||||
1 if workdir => {
|
||||
let (error, text) = output(&request, "work-read").unwrap();
|
||||
assert!(
|
||||
!error && text.contains("clean"),
|
||||
"real background Read: {text}"
|
||||
);
|
||||
call(
|
||||
"work-write",
|
||||
"Write",
|
||||
json!({"file_path":"README.md","content":"background changed after real Read\n"}),
|
||||
)
|
||||
}
|
||||
1 | 2 if phase == if workdir { 2 } else { 1 } => {
|
||||
let (error, text) =
|
||||
output(&request, if workdir { "work-write" } else { "work-query" })
|
||||
.unwrap();
|
||||
assert!(!error, "bounded domain/file work: {text}");
|
||||
call("reply-origin", "WorkerList", json!({}))
|
||||
}
|
||||
2 | 3 if phase == if workdir { 3 } else { 2 } => {
|
||||
let list = value(&request, "reply-origin");
|
||||
let origins = list["items"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|item| item["relation"] == "reply")
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(origins.len(), 1, "confirmed origin: {list}");
|
||||
assert_eq!(origins[0]["permissions"], json!(["notify"]));
|
||||
let c = self.context.lock().unwrap();
|
||||
call(
|
||||
"report",
|
||||
"WorkerNotify",
|
||||
json!({"subject":origins[0]["subject"],"content":format!("{REPORT}; conversation={}; work={}", c.a, if workdir {"README read/write"} else {"bounded objective query"})}),
|
||||
)
|
||||
}
|
||||
3 | 4 => {
|
||||
assert_eq!(value(&request, "report")["state"], "accepted");
|
||||
done()
|
||||
}
|
||||
_ => panic!("unexpected background model phase {phase}"),
|
||||
}
|
||||
} else {
|
||||
assert!(
|
||||
!request.tools.iter().any(|tool| matches!(
|
||||
tool.name.as_str(),
|
||||
"Bash" | "Read" | "Write" | "Edit" | "SubWorkerSpawn"
|
||||
)),
|
||||
"actual async-chat must stay restricted"
|
||||
);
|
||||
let phase = self.body_phase.fetch_add(1, Ordering::SeqCst);
|
||||
match phase {
|
||||
0 => call(
|
||||
"origin-read",
|
||||
"ConversationRead",
|
||||
json!({"conversation_id":self.context.lock().unwrap().a}),
|
||||
),
|
||||
1 => call(
|
||||
"spawn-first",
|
||||
"WorkerSpawn",
|
||||
self.spawn(&value(&request, "origin-read")),
|
||||
),
|
||||
2 => {
|
||||
let (error, text) = output(&request, "spawn-first").unwrap();
|
||||
if self.lose_spawn_ack {
|
||||
assert!(
|
||||
error && text.contains("unknown"),
|
||||
"lost ACK must not claim acceptance/rejection: {text}"
|
||||
);
|
||||
} else {
|
||||
assert!(!error, "first spawn: {text}");
|
||||
}
|
||||
call(
|
||||
"spawn-retry",
|
||||
"WorkerSpawn",
|
||||
self.spawn(&value(&request, "origin-read")),
|
||||
)
|
||||
}
|
||||
3 => {
|
||||
let retry = value(&request, "spawn-retry");
|
||||
if !self.lose_spawn_ack {
|
||||
assert_eq!(
|
||||
value(&request, "spawn-first")["worker_id"],
|
||||
retry["worker_id"]
|
||||
);
|
||||
}
|
||||
call(
|
||||
"spawn-exact-replay",
|
||||
"WorkerSpawn",
|
||||
self.spawn(&value(&request, "origin-read")),
|
||||
)
|
||||
}
|
||||
4 => {
|
||||
let child = value(&request, "spawn-retry");
|
||||
assert_eq!(
|
||||
value(&request, "spawn-exact-replay")["worker_id"],
|
||||
child["worker_id"]
|
||||
);
|
||||
self.context.lock().unwrap().spawned = Some(RuntimeWorkerRef::new(
|
||||
child["runtime_id"].as_str().unwrap(),
|
||||
child["worker_id"].as_str().unwrap(),
|
||||
));
|
||||
done()
|
||||
}
|
||||
5 => call(
|
||||
"b-read",
|
||||
"ConversationRead",
|
||||
json!({"conversation_id":self.context.lock().unwrap().b}),
|
||||
),
|
||||
6 => call(
|
||||
"b-send",
|
||||
"ConversationSend",
|
||||
self.send(&request, "b-read", PUBLIC_B, "answer-b", false),
|
||||
),
|
||||
7 => {
|
||||
value(&request, "b-send");
|
||||
done()
|
||||
}
|
||||
8 => {
|
||||
assert!(request.items.iter().any(|item| matches!(item, Item::Message { content, .. } if serde_json::to_string(content).unwrap().contains(REPORT))), "ordinary report must reach model as private notification");
|
||||
call(
|
||||
"a-reread",
|
||||
"ConversationRead",
|
||||
json!({"conversation_id":self.context.lock().unwrap().a}),
|
||||
)
|
||||
}
|
||||
9 => {
|
||||
self.private_report.try_send(()).unwrap();
|
||||
tokio::time::timeout(
|
||||
std::time::Duration::from_secs(60),
|
||||
self.release_send.acquire(),
|
||||
)
|
||||
.await
|
||||
.expect("inspect private report before explicit publication")
|
||||
.unwrap()
|
||||
.forget();
|
||||
call(
|
||||
"a-send",
|
||||
"ConversationSend",
|
||||
self.send(&request, "a-reread", PUBLIC_A, "answer-a", true),
|
||||
)
|
||||
}
|
||||
10 => {
|
||||
value(&request, "a-send");
|
||||
self.completed.try_send(()).unwrap();
|
||||
done()
|
||||
}
|
||||
11 => {
|
||||
let child = value(&request, "spawn-retry");
|
||||
call(
|
||||
"revoked-notify",
|
||||
"WorkerNotify",
|
||||
json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"content":"MUST_NOT_REACH_REVOKED_CHILD"}),
|
||||
)
|
||||
}
|
||||
12 => {
|
||||
let (error, text) = output(&request, "revoked-notify").unwrap();
|
||||
assert!(
|
||||
error,
|
||||
"revocation must be a public Tool refusal, never success: {text}"
|
||||
);
|
||||
self.completed.try_send(()).unwrap();
|
||||
done()
|
||||
}
|
||||
_ => panic!("unexpected chat model phase {phase}"),
|
||||
}
|
||||
};
|
||||
Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok))))
|
||||
}
|
||||
}
|
||||
|
||||
fn publish_profiles(api: &WorkspaceApi) {
|
||||
let state = api
|
||||
.config_store
|
||||
.load_workspace_config(TEST_WORKSPACE_ID)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let main = state
|
||||
.snapshot
|
||||
.get(&config_source::VirtualPath::parse("main.dcdl").unwrap())
|
||||
.unwrap();
|
||||
let content = r#"{ profile = { default_profile = "project:async-chat"; entries = [
|
||||
{ selector = "project:async-chat"; label = "Async chat"; description = "Unmodified built-in chat";
|
||||
profile = import "$builtin/profiles/async-chat.dcdl"; },
|
||||
{ selector = "project:chat-background"; label = "Background work"; description = "Bounded file or domain work";
|
||||
profile = (import "$builtin/profiles/companion.dcdl") // { slug = "chat-background"; scope = "workspace_write";
|
||||
feature = { subjektiv = { enabled = false; }; worker = { enabled = true; direct_spawn = false; }; };
|
||||
permissions = { default_action = "deny"; rule = [
|
||||
{ tool = "Read"; pattern = "*"; action = "allow"; },
|
||||
{ tool = "Write"; pattern = "*"; action = "allow"; },
|
||||
{ tool = "QueryObjective"; pattern = "*"; action = "allow"; },
|
||||
{ tool = "WorkerList"; pattern = "*"; action = "allow"; },
|
||||
{ tool = "WorkerNotify"; pattern = "*"; action = "allow"; }
|
||||
]; };
|
||||
}; }
|
||||
]; }; }"#;
|
||||
commit_workspace_config_tree(
|
||||
api,
|
||||
TEST_WORKSPACE_ID,
|
||||
&ConfigCommitRequest {
|
||||
base_digest: state.snapshot.digest.clone(),
|
||||
entrypoints: state.contract.entrypoints.clone(),
|
||||
changes: vec![config_source::ConfigTreeChange::Update {
|
||||
path: main.path.clone(),
|
||||
expected_digest: main.content_digest.clone(),
|
||||
content: content.into(),
|
||||
}],
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
async fn account_request(
|
||||
h: &ConnectedWorkerHarness,
|
||||
method: &str,
|
||||
path: &str,
|
||||
body: Option<Value>,
|
||||
) -> Value {
|
||||
let (status, result) = response(&h.api, human_request(&h.owner, method, path, body)).await;
|
||||
assert!(status.is_success(), "{method} {path}: {status}: {result}");
|
||||
result
|
||||
}
|
||||
fn messages(conversation: &str) -> String {
|
||||
format!("/api/w/{TEST_WORKSPACE_ID}/chat/conversations/{conversation}/messages")
|
||||
}
|
||||
async fn idle(
|
||||
events: &mut tokio::sync::broadcast::Receiver<
|
||||
worker_runtime::observation::WorkerObservationEvent,
|
||||
>,
|
||||
worker: &str,
|
||||
) {
|
||||
tokio::time::timeout(std::time::Duration::from_secs(60),async {
|
||||
loop {
|
||||
let event = events.recv().await.unwrap();
|
||||
if event.worker_ref.worker_id.to_string() == worker && matches!(event.payload,protocol::Event::WorkerState {snapshot} if snapshot.state == protocol::WorkerState::Idle) {break;}
|
||||
}
|
||||
}).await.expect("real chat Controller Idle boundary");
|
||||
}
|
||||
async fn connected_chat_delegation(explicit_workdir: bool, lose_spawn_ack: bool) {
|
||||
let context = Arc::new(Mutex::new(Context::default()));
|
||||
let (started, mut start) = tokio::sync::mpsc::channel(1);
|
||||
let (completed, mut completion) = tokio::sync::mpsc::channel(2);
|
||||
let (private_report, mut private_input) = tokio::sync::mpsc::channel(1);
|
||||
let release_send = Arc::new(tokio::sync::Semaphore::new(0));
|
||||
let release = Arc::new(tokio::sync::Semaphore::new(0));
|
||||
let model = DelegationModel {
|
||||
context: context.clone(),
|
||||
body_phase: Arc::new(AtomicUsize::new(0)),
|
||||
child_phase: Arc::new(AtomicUsize::new(0)),
|
||||
child_started: started,
|
||||
release_work: release.clone(),
|
||||
private_report,
|
||||
release_send: release_send.clone(),
|
||||
completed,
|
||||
lose_spawn_ack,
|
||||
};
|
||||
let body_phase = model.body_phase.clone();
|
||||
let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
let dropped_response = dropped.clone();
|
||||
let h = ConnectedWorkerHarness::start_with_layer(model,move |router| router.layer(axum::middleware::from_fn(
|
||||
move |request:axum::extract::Request,next:axum::middleware::Next| {
|
||||
let dropped = dropped_response.clone();
|
||||
async move {
|
||||
let spawn = request.method() == axum::http::Method::POST && request.uri().path() == format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers");
|
||||
let response = next.run(request).await;
|
||||
if spawn && response.status().is_success() && lose_spawn_ack && !dropped.swap(true,Ordering::SeqCst) {
|
||||
// Discard only the response after real registry/grant persistence.
|
||||
(StatusCode::GATEWAY_TIMEOUT,Json(json!({"error":"Gateway Timeout","message":"spawn outcome unknown; retry original operation and input","diagnostics":[]}))).into_response()
|
||||
} else {response}
|
||||
}
|
||||
}
|
||||
))).await;
|
||||
publish_profiles(&h.api);
|
||||
let mut child_permissions = vec![
|
||||
operation("worker.messaging", false),
|
||||
operation("worker.list", true),
|
||||
];
|
||||
let mut readme = None;
|
||||
if explicit_workdir {
|
||||
let created = account_request(&h,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"),Some(json!({"runtime_id":CONNECTED_RUNTIME,"repository_key":"test-repository","selector":"HEAD","operation_id":"chat-workdir"}))).await;
|
||||
let id = created["item"]["working_directory_id"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_owned();
|
||||
readme = Some(
|
||||
h.materializer
|
||||
.bind_working_directory(&id, None)
|
||||
.unwrap()
|
||||
.root()
|
||||
.join("README.md"),
|
||||
);
|
||||
for operation in ["filesystem.read", "filesystem.write"] {
|
||||
child_permissions.push(ScopedPermission {
|
||||
operation: operation.into(),
|
||||
resource: PermissionResource::Workdir {
|
||||
working_directory_id: id.clone(),
|
||||
path: ".".into(),
|
||||
recursive: true,
|
||||
},
|
||||
});
|
||||
}
|
||||
context.lock().unwrap().workdir = Some(id);
|
||||
} else {
|
||||
child_permissions.push(operation("objective.query", true));
|
||||
}
|
||||
context.lock().unwrap().permissions = child_permissions.clone();
|
||||
let store = open_subjektiv_store(&h.api).unwrap();
|
||||
let subject = store
|
||||
.create_subject(crate::subjektiv::SubjectRole::new("async-chat-delegation").unwrap())
|
||||
.unwrap()
|
||||
.id;
|
||||
let mut execution = [
|
||||
"session.record",
|
||||
"memory.resident.read",
|
||||
"memory.resident.supply",
|
||||
"journal.startup.read",
|
||||
"journal.startup.supply",
|
||||
"chat.read",
|
||||
"chat.send",
|
||||
]
|
||||
.into_iter()
|
||||
.map(|op| subject_permission(&subject, op))
|
||||
.collect::<Vec<_>>();
|
||||
execution.extend([
|
||||
operation("worker.create", true),
|
||||
operation("worker.messaging", false),
|
||||
]);
|
||||
account_request(&h,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"),Some(json!({"operation_id":"chat-policy","expected_change_id":null,"execution":execution,"delegation":child_permissions}))).await;
|
||||
let original_policy = store.subject_permissions(&subject).unwrap().unwrap();
|
||||
for (name, target) in [("conversation-a", true), ("conversation-b", false)] {
|
||||
let c = account_request(
|
||||
&h,
|
||||
"POST",
|
||||
&format!("/api/w/{TEST_WORKSPACE_ID}/chat/conversations"),
|
||||
Some(json!({"operation_id":name,"subject_id":subject})),
|
||||
)
|
||||
.await;
|
||||
let mut context = context.lock().unwrap();
|
||||
if target {
|
||||
context.a = c["id"].as_str().unwrap().into();
|
||||
} else {
|
||||
context.b = c["id"].as_str().unwrap().into();
|
||||
}
|
||||
}
|
||||
let (a, b) = {
|
||||
let c = context.lock().unwrap();
|
||||
(c.a.clone(), c.b.clone())
|
||||
};
|
||||
let source = account_request(&h,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/workers"),Some(json!({"runtime_id":CONNECTED_RUNTIME,"display_name":"Restricted async chat","profile":"project:async-chat","initial_submit":[],"workdir_attachments":[],"feature_connections":{"subjektiv":{"subject_id":subject}}}))).await;
|
||||
let source = RuntimeWorkerRef::new(
|
||||
source["runtime_id"].as_str().unwrap(),
|
||||
source["worker_id"].as_str().unwrap(),
|
||||
);
|
||||
let mut events = h.runtime.subscribe_worker_observation().unwrap();
|
||||
let first = account_request(&h,"POST",&messages(&a),Some(json!({"operation_id":"request-a","text":"Please perform bounded background work for A"}))).await;
|
||||
assert_eq!(first["delivery"]["state"], "accepted");
|
||||
tokio::time::timeout(std::time::Duration::from_secs(60), start.recv())
|
||||
.await
|
||||
.expect("real child started")
|
||||
.unwrap();
|
||||
idle(&mut events, &source.worker_id).await;
|
||||
assert_eq!(
|
||||
body_phase.load(Ordering::SeqCst),
|
||||
5,
|
||||
"body finishes rather than polling/waiting for work"
|
||||
);
|
||||
assert_eq!(
|
||||
h.runtime.list_workers().unwrap().len(),
|
||||
2,
|
||||
"same public Spawn operation retains one child"
|
||||
);
|
||||
if let Some(path) = &readme {
|
||||
assert_eq!(fs::read_to_string(path).unwrap(), "clean\n");
|
||||
}
|
||||
assert_eq!(
|
||||
account_request(&h, "GET", &messages(&a), None).await["messages"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.len(),
|
||||
1,
|
||||
"creation/assistant output does not publish a result"
|
||||
);
|
||||
let second = account_request(&h,"POST",&messages(&b),Some(json!({"operation_id":"request-b","text":"A separate question B while background work continues"}))).await;
|
||||
assert_eq!(second["delivery"]["state"], "accepted");
|
||||
idle(&mut events, &source.worker_id).await;
|
||||
let page_b = account_request(&h, "GET", &messages(&b), None).await;
|
||||
assert_eq!(page_b["messages"].as_array().unwrap().len(), 2);
|
||||
assert_eq!(page_b["messages"][1]["text"], PUBLIC_B);
|
||||
assert_eq!(page_b["messages"][1]["reply_to"], second["message"]["id"]);
|
||||
assert_eq!(
|
||||
account_request(&h, "GET", &messages(&a), None).await["messages"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.len(),
|
||||
1
|
||||
);
|
||||
release.add_permits(1);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(60), private_input.recv())
|
||||
.await
|
||||
.expect("private report model boundary")
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
account_request(&h, "GET", &messages(&a), None).await["messages"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.len(),
|
||||
1,
|
||||
"ordinary WorkerNotify is not public chat output"
|
||||
);
|
||||
let source_snapshot = h
|
||||
.factory
|
||||
.retained_session_snapshot(&worker_runtime::identity::WorkerRef::try_from(&source).unwrap())
|
||||
.unwrap();
|
||||
assert_eq!(source_snapshot.snapshot.entries.iter().filter(|entry| matches!(&entry.data, protocol::SessionSnapshotEntryData::SystemItem {item_kind,content,..} if item_kind == "notification" && content.contains(REPORT))).count(),1,"ordinary report committed exactly once before model supply");
|
||||
if let Some(path) = &readme {
|
||||
assert_eq!(
|
||||
fs::read_to_string(path).unwrap(),
|
||||
"background changed after real Read\n"
|
||||
);
|
||||
}
|
||||
release_send.add_permits(1);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(60), completion.recv())
|
||||
.await
|
||||
.expect("ordinary report and explicit A send")
|
||||
.unwrap();
|
||||
idle(&mut events, &source.worker_id).await;
|
||||
let page_a = account_request(&h, "GET", &messages(&a), None).await;
|
||||
assert_eq!(page_a["messages"].as_array().unwrap().len(), 2);
|
||||
assert_eq!(page_a["messages"][1]["text"], PUBLIC_A);
|
||||
assert_eq!(page_a["messages"][1]["reply_to"], first["message"]["id"]);
|
||||
assert_eq!(
|
||||
account_request(&h, "GET", &messages(&b), None).await,
|
||||
page_b,
|
||||
"A's report never crosses into B"
|
||||
);
|
||||
assert!(!page_a.to_string().contains(REPORT) && !page_a.to_string().contains("PRIVATE_CHAT"));
|
||||
if let Some(path) = &readme {
|
||||
assert_eq!(
|
||||
fs::read_to_string(path).unwrap(),
|
||||
"background changed after real Read\n"
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
store.subject_permissions(&subject).unwrap().unwrap(),
|
||||
original_policy,
|
||||
"no per-target Subject policy copy"
|
||||
);
|
||||
let child = context.lock().unwrap().spawned.clone().unwrap();
|
||||
let forward = h
|
||||
.api
|
||||
.store
|
||||
.get_active_worker_control_grant(TEST_WORKSPACE_ID, &source, &child)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let reverse = h
|
||||
.api
|
||||
.store
|
||||
.get_active_worker_control_grant(TEST_WORKSPACE_ID, &child, &source)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(reverse.permissions, vec!["notify".to_owned()]);
|
||||
assert!(
|
||||
h.api
|
||||
.store
|
||||
.require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &child)
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
let grant = store
|
||||
.worker_execution_grant(&child.runtime_id, &child.worker_id)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(grant.permissions, context.lock().unwrap().permissions);
|
||||
assert!(
|
||||
!grant
|
||||
.permissions
|
||||
.iter()
|
||||
.any(|permission| permission.operation == "chat.send")
|
||||
);
|
||||
assert!(
|
||||
h.api
|
||||
.store
|
||||
.list_worker_workdir_links(TEST_WORKSPACE_ID, &source)
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
if explicit_workdir {
|
||||
h.api
|
||||
.store
|
||||
.revoke_worker_control_grant(
|
||||
TEST_WORKSPACE_ID,
|
||||
&forward.grant_id,
|
||||
&now_registry_timestamp(),
|
||||
)
|
||||
.unwrap();
|
||||
} else {
|
||||
let narrowed = execution
|
||||
.into_iter()
|
||||
.filter(|permission| permission.operation != "worker.messaging")
|
||||
.collect::<Vec<_>>();
|
||||
account_request(&h,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"),Some(json!({"operation_id":"revoke-messaging","expected_change_id":original_policy.change_id,"execution":narrowed,"delegation":context.lock().unwrap().permissions}))).await;
|
||||
}
|
||||
account_request(&h,"POST",&messages(&a),Some(json!({"operation_id":"request-denied","text":"Attempt follow-up after authorization is revoked"}))).await;
|
||||
tokio::time::timeout(std::time::Duration::from_secs(60), completion.recv())
|
||||
.await
|
||||
.expect("public refusal after revocation")
|
||||
.unwrap();
|
||||
idle(&mut events, &source.worker_id).await;
|
||||
let snapshot = h
|
||||
.factory
|
||||
.retained_session_snapshot(&worker_runtime::identity::WorkerRef::try_from(&child).unwrap())
|
||||
.unwrap();
|
||||
assert!(
|
||||
!serde_json::to_string(&snapshot.snapshot)
|
||||
.unwrap()
|
||||
.contains("MUST_NOT_REACH_REVOKED_CHILD")
|
||||
);
|
||||
assert_eq!(
|
||||
account_request(&h, "GET", &messages(&a), None).await["messages"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.len(),
|
||||
3,
|
||||
"failed Notify creates no automatic success publication"
|
||||
);
|
||||
assert_eq!(dropped.load(Ordering::SeqCst), lose_spawn_ack);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
|
||||
async fn restricted_chat_delegates_real_file_work_answers_b_then_explicitly_returns_to_a() {
|
||||
connected_chat_delegation(true, true).await;
|
||||
}
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
|
||||
async fn restricted_workdirless_chat_delegates_bounded_query_and_refuses_revoked_messaging() {
|
||||
connected_chat_delegation(false, false).await;
|
||||
}
|
||||
@@ -5,6 +5,8 @@
|
||||
//! runs a real ProfileRuntimeWorkerFactory/Controller and materialized Workdir;
|
||||
//! only its external model stream is scripted. Neither fixture claims OS isolation.
|
||||
use super::*;
|
||||
#[path = "chat_delegation_connected_tests.rs"]
|
||||
mod chat_delegation_connected;
|
||||
#[path = "chat_tests.rs"]
|
||||
mod chat_tests;
|
||||
#[path = "subject_job_model_supply_tests.rs"]
|
||||
|
||||
Reference in New Issue
Block a user