Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

Merge remote-tracking branch 'origin/develop' into work/t-533-plugin-worker-composed

Browse Source 
# Conflicts:
#	crates/server-api/src/lib.rs
#	crates/server-api/src/plugin.rs
#	crates/server-api/src/plugin_types.rs
#	crates/worker-runtime/Cargo.toml
#	crates/worker-runtime/src/workspace_request/plugin_delivery.rs
#	crates/workspace-server/Cargo.toml
#	crates/workspace-server/src/lib.rs
#	crates/workspace-server/src/server.rs
#	crates/workspace-server/src/server/tests/subject_authorization_tests.rs
#	crates/workspace-server/src/server_plugin_delivery.rs
#	crates/workspace-server/src/server_plugin_delivery_tests.rs
#	openapi/server-api.json
#	web/workspace/src/lib/generated/companion-api.ts
#	web/workspace/src/lib/generated/plugin-api.ts
#	web/workspace/src/lib/generated/repository-api.ts
#	web/workspace/src/lib/generated/runtime-api.ts
#	web/workspace/src/lib/generated/ticket-api.ts
#	web/workspace/src/lib/generated/workdir-api.ts
#	web/workspace/src/lib/generated/worker-launch-api.ts
This commit is contained in:
Hare
2026-10-30 07:20:50 +09:00
parent b824195b36 318c1a9ce9
commit 94d458883e
91 changed files with 12098 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+24 -1
View File
@@ -546,6 +546,7 @@ fn durable_parent_notification_target<St: Store + Clone + Send + Sync + 'static>
) -> crate::spawn::tool::ParentNotificationTarget {
crate::spawn::tool::ParentNotificationTarget::Durable(Arc::new(move |method| {
let Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
@@ -554,6 +555,7 @@ fn durable_parent_notification_target<St: Store + Clone + Send + Sync + 'static>
return;
};
let (source_namespace, provenance) = resolved_input_source(&pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
match pending_submissions.accept_notification_from_source(
notification_request_id,
message,
@@ -1499,7 +1501,7 @@ where
// Keep the alias-routed schemas installed for the Worker lifetime. The
// attachment set may be empty or mutate after registration, so target
// resolution and provider capability checks happen per invocation.
let workdir_tools = if restrict_workdir_tools_to_current_capabilities {
let mut workdir_tools = if restrict_workdir_tools_to_current_capabilities {
tools::routed_builtin_tools_for_current_capabilities(
workdir_sessions.clone(),
tracker.clone(),
@@ -1512,6 +1514,11 @@ where
bash_output.clone(),
)
};
if let Some(policy) = &spawner_manifest.permissions {
workdir_tools.retain(|definition| {
!crate::permission::tool_is_unconditionally_denied(policy, &definition().0.name)
});
}
worker.engine_mut().register_tools(workdir_tools);
}
if feature_config.image.enabled && model_supports_image_attachments(&spawner_manifest.model) {
@@ -1658,6 +1665,15 @@ where
}
feature_prompt_contribution = Some((Some(resident_summary_source), None));
}
if backend_job.is_none()
&& let Some(chat) =
crate::feature::builtin::subjektiv_chat::SubjektivChatFeature::from_resolved_config(
&worker.manifest().feature.subjektiv,
subjektiv_host.clone(),
)?
{
feature_registry.add_module(chat);
}
if ordinary_subjektiv_features_enabled
&& let Some(subjektiv_memory) =
crate::feature::builtin::subjektiv_memory::SubjektivMemoryFeature::from_resolved_config(
@@ -2275,12 +2291,14 @@ where
None
}
Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
} => {
let (source_namespace, provenance) =
resolved_input_source(pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
durably_accept_notification(
notification_request_id,
message,
@@ -2756,12 +2774,14 @@ async fn controller_loop<C, St>(
}
Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
} => {
let (source_namespace, provenance) =
resolved_input_source(&pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
durably_accept_notification(
notification_request_id,
message,
@@ -3678,12 +3698,14 @@ where
});
}
Some(Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
}) => {
let (source_namespace, provenance) =
resolved_input_source(pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
if durably_accept_notification(
notification_request_id,
message,
@@ -4907,6 +4929,7 @@ mod tests {
let pending = PendingSubmissionHandle::for_test(dir.path());
let (event_tx, mut event_rx) = broadcast::channel(4);
let method = Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "completion-1".into(),
message: "subworker completed".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire,
crates/worker/src/feature/builtin.rs
+1
View File
@@ -36,6 +36,7 @@ pub mod orchestration;
mod resource_projection;
pub mod session_explore;
pub(crate) mod sub_worker;
pub(crate) mod subjektiv_chat;
pub(crate) mod subjektiv_memory;
pub(crate) mod subjektiv_session;
pub mod task;
crates/worker/src/feature/builtin/manage_worker.rs
+1
View File
@@ -210,6 +210,7 @@ impl WorkerControlService for WorkspaceWorkerControlService {
runtime_id,
worker_id,
request: server_api::RuntimeWorkerInputRequest {
submission_request_id: None,
kind: Some(kind.to_string()),
content,
segments: None,
crates/worker/src/feature/builtin/subjektiv_chat.rs
+222
View File
@@ -0,0 +1,222 @@
//! Explicit public conversation access for the connected Subject body.
//!
//! This module installs no output or lifecycle hooks. Internal assistant text,
//! WorkerNotify reports and turn completion never become public messages here.
//! Current body ownership, conversation access and observed-input/idempotency checks
//! remain Backend-owned; the model supplies neither sender nor Subject identity.
use std::sync::Arc;
use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
use async_trait::async_trait;
use schemars::JsonSchema;
use serde::Deserialize;
use server_api::{
ChatMessagePageQuery, ChatSendMessageRequest, ChatWorkerOperation, ChatWorkerResponse,
};
use crate::feature::{
FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution,
ToolDeclaration,
};
use crate::subjektiv::SubjektivHostConnection;
const READ_TOOL: &str = "ConversationRead";
const SEND_TOOL: &str = "ConversationSend";
const READ_DESCRIPTION: &str = "Read a bounded page of committed text-only messages in an authorized conversation for the explicitly connected Subject. Returns the current last_user_message_id; use after_sequence for continuation. This reads public messages, not internal Worker logs or notification-delivery evidence.";
const SEND_DESCRIPTION: &str = "Explicitly send one text-only public message as the connected Subject to an authorized conversation. Use a stable operation_id and the observed expected_last_user_message_id. Sender identity is Host-bound. Exact retries retain the full request and operation_id; conflicts require rereading and reconsideration, while a lost response is not proof of failure. This is the only operation here that publishes assistant text; turn completion and background WorkerNotify reports do not send messages. No attachments or file-path attachment substitution.";
const MAX_TOOL_OUTPUT_BYTES: usize = 128 * 1024;
#[derive(Clone)]
pub(crate) struct SubjektivChatFeature {
host: SubjektivHostConnection,
}
impl SubjektivChatFeature {
pub(crate) fn from_resolved_config(
config: &manifest::ResolvedSubjektivFeatureConfig,
host: Option<SubjektivHostConnection>,
) -> std::io::Result<Option<Self>> {
if !config.profile.enabled || !config.profile.chat.enabled {
return Ok(None);
}
config.validate_execution().map_err(std::io::Error::other)?;
let host = host.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"subjektiv chat requires an explicit live Subject Host connection",
)
})?;
if config.profile.consolidation_tools {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"subjektiv chat is a body capability, not a consolidation Job capability",
));
}
Ok(Some(Self { host }))
}
}
impl FeatureModule for SubjektivChatFeature {
fn descriptor(&self) -> FeatureDescriptor {
FeatureDescriptor::builtin("subjektiv-chat", "Subject Conversation")
.with_description("Explicit text-only public conversation read and send operations.")
.with_tool(ToolDeclaration::new(READ_TOOL, READ_DESCRIPTION))
.with_tool(ToolDeclaration::new(SEND_TOOL, SEND_DESCRIPTION))
}
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
for operation in [Operation::Read, Operation::Send] {
context.tools().register(ToolContribution::new(
operation.name(),
tool_definition(self.host.clone(), operation),
))?;
}
Ok(())
}
}
#[derive(Clone, Copy)]
enum Operation {
Read,
Send,
}
impl Operation {
fn name(self) -> &'static str {
match self {
Self::Read => READ_TOOL,
Self::Send => SEND_TOOL,
}
}
fn description(self) -> &'static str {
match self {
Self::Read => READ_DESCRIPTION,
Self::Send => SEND_DESCRIPTION,
}
}
}
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct ReadParams {
conversation_id: String,
after_sequence: Option<u64>,
limit: Option<u32>,
}
fn schema<T: JsonSchema>() -> serde_json::Value {
serde_json::to_value(schemars::schema_for!(T)).expect("serializable chat Tool schema")
}
fn tool_definition(host: SubjektivHostConnection, operation: Operation) -> ToolDefinition {
Arc::new(move || {
let schema = match operation {
Operation::Read => schema::<ReadParams>(),
Operation::Send => schema::<ChatSendMessageRequest>(),
};
let tool: Arc<dyn Tool> = Arc::new(ConversationTool {
host: host.clone(),
operation,
});
(
ToolMeta::new(operation.name())
.description(operation.description())
.input_schema(schema),
tool,
)
})
}
struct ConversationTool {
host: SubjektivHostConnection,
operation: Operation,
}
fn parse<T: serde::de::DeserializeOwned>(input: &str) -> Result<T, ToolError> {
serde_json::from_str(input).map_err(|error| ToolError::InvalidArgument(error.to_string()))
}
#[async_trait]
impl Tool for ConversationTool {
async fn execute(
&self,
input_json: &str,
_context: ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let operation = match self.operation {
Operation::Read => {
let input: ReadParams = parse(input_json)?;
if input.conversation_id.trim().is_empty()
|| input.limit.is_some_and(|limit| {
!(1..=server_api::CHAT_MAX_PAGE_LIMIT).contains(&limit)
})
{
return Err(ToolError::InvalidArgument(
"conversation_id is required and limit must be within 1..=100".into(),
));
}
ChatWorkerOperation::Read {
conversation_id: input.conversation_id,
query: ChatMessagePageQuery {
after_sequence: input.after_sequence,
// Small default pages keep the chat body responsive; callers
// can explicitly request more within the Backend ceiling.
limit: Some(input.limit.unwrap_or(5)),
},
}
}
Operation::Send => {
let request: ChatSendMessageRequest = parse(input_json)?;
if request.conversation_id.trim().is_empty()
|| request.operation_id.trim().is_empty()
|| request.text.trim().is_empty()
|| request.text.len() > server_api::CHAT_MAX_TEXT_BYTES
{
return Err(ToolError::InvalidArgument(
"conversation_id, operation_id and nonblank text are required; text exceeds the limit if larger than 16 KiB".into(),
));
}
ChatWorkerOperation::Send { request }
}
};
let response = self.host.chat(operation).map_err(ToolError::from)?;
let (summary, value) = match (self.operation, response) {
(Operation::Read, ChatWorkerResponse::Messages { page }) => (
format!(
"Read {} committed conversation message(s).",
page.messages.len()
),
serde_json::to_value(page),
),
(Operation::Send, ChatWorkerResponse::Sent { message }) => (
"Committed conversation message.".to_string(),
serde_json::to_value(message),
),
_ => {
return Err(ToolError::ExecutionFailed(
"unexpected chat Backend response".into(),
));
}
};
let content = serde_json::to_string(
&value.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?,
)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
if content.len() > MAX_TOOL_OUTPUT_BYTES {
return Err(ToolError::ExecutionFailed(
"conversation page exceeds the model-visible output budget; read a smaller page"
.into(),
));
}
Ok(ToolOutput {
summary,
content: Some(content),
attachments: Vec::new(),
})
}
}
#[cfg(test)]
mod tests;
crates/worker/src/feature/builtin/subjektiv_chat/tests.rs
+465
View File
@@ -0,0 +1,465 @@
use super::*;
use crate::worker::{WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceResponse};
use server_api::{
ChatConversation, ChatMessage, ChatMessagePage, ChatMessageSender, ChatWorkerRequest,
};
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU16, AtomicUsize, Ordering};
#[derive(Debug)]
struct ChatBackendFixture {
allowed: Mutex<Vec<&'static str>>,
requests: Mutex<Vec<WorkspaceRequest>>,
status: AtomicU16,
unavailable: AtomicBool,
wrong_variant: AtomicBool,
page_text_bytes: AtomicUsize,
}
impl Default for ChatBackendFixture {
fn default() -> Self {
Self {
allowed: Mutex::new(vec!["chat.read", "chat.send"]),
requests: Mutex::new(Vec::new()),
status: AtomicU16::new(200),
unavailable: AtomicBool::new(false),
wrong_variant: AtomicBool::new(false),
page_text_bytes: AtomicUsize::new(0),
}
}
}
fn message(request: ChatSendMessageRequest) -> ChatMessage {
ChatMessage {
id: "message-committed".into(),
conversation_id: request.conversation_id,
sequence: 7,
sender: ChatMessageSender::Subject {
subject_id: "subject-fixture".into(),
},
text: request.text,
reply_to: request.reply_to,
work_ref: request.work_ref,
operation_id: request.operation_id,
last_user_message_id: request.expected_last_user_message_id,
}
}
fn send_input() -> serde_json::Value {
serde_json::json!({
"conversation_id": "conversation-1",
"operation_id": "stable-operation-1",
"text": "I evaluated the background report.",
"reply_to": "user-message-1",
"work_ref": "T-766",
"expected_last_user_message_id": "user-message-3"
})
}
impl WorkspaceClient for ChatBackendFixture {
fn workspace_id(&self) -> Option<&str> {
Some("workspace-chat")
}
fn kind(&self) -> &str {
"chat-test"
}
fn is_available(&self) -> bool {
true
}
fn execute(
&self,
request: WorkspaceRequest,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
self.requests.lock().unwrap().push(request.clone());
if let Some(response) = crate::subjektiv::test_subject_authorization_response(
&request,
"workspace-chat",
&self.allowed.lock().unwrap(),
) {
return Ok(response);
}
assert_eq!(request.path, "/api/w/workspace-chat/subjektiv/chat/worker");
assert_eq!(request.method, crate::WorkspaceRequestMethod::Post);
if self.unavailable.load(Ordering::SeqCst) {
return Err(WorkspaceClientError::Unavailable(
"response lost; acceptance unknown".into(),
));
}
let status = self.status.load(Ordering::SeqCst);
if status != 200 {
return Ok(WorkspaceResponse {
status,
body: serde_json::json!({"message":"Backend rejected chat operation"}).to_string(),
});
}
let input: ChatWorkerRequest =
serde_json::from_str(request.body.as_deref().unwrap()).unwrap();
let response = match input.operation {
ChatWorkerOperation::Send { request } if !self.wrong_variant.load(Ordering::SeqCst) => {
ChatWorkerResponse::Sent {
message: message(request),
}
}
_ => ChatWorkerResponse::Messages {
page: ChatMessagePage {
conversation: ChatConversation {
id: "conversation-1".into(),
subject_id: "subject-fixture".into(),
participant_account_id: "account-fixture".into(),
last_user_message_id: Some("user-message-3".into()),
},
messages: if self.page_text_bytes.load(Ordering::SeqCst) > 0 {
let mut request: ChatSendMessageRequest =
serde_json::from_value(send_input()).unwrap();
request.text = "x".repeat(self.page_text_bytes.load(Ordering::SeqCst));
vec![message(request)]
} else {
Vec::new()
},
has_more: false,
},
},
};
Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(&response).unwrap(),
})
}
}
impl ChatBackendFixture {
fn chat_requests(&self) -> Vec<ChatWorkerRequest> {
self.requests
.lock()
.unwrap()
.iter()
.filter(|request| request.path.ends_with("/chat/worker"))
.map(|request| serde_json::from_str(request.body.as_deref().unwrap()).unwrap())
.collect()
}
}
fn tool(client: Arc<ChatBackendFixture>, operation: Operation) -> ConversationTool {
ConversationTool {
host: crate::subjektiv::test_connection(client),
operation,
}
}
async fn invoke(
tool: &ConversationTool,
input: serde_json::Value,
) -> Result<ToolOutput, ToolError> {
tool.execute(
&input.to_string(),
ToolExecutionContext::new("call-chat", "batch", 0),
)
.await
}
#[test]
fn feature_is_opt_in_and_requires_explicit_connection_without_hooks() {
let mut config = manifest::ResolvedSubjektivFeatureConfig::default();
config.profile.enabled = true;
assert!(
SubjektivChatFeature::from_resolved_config(&config, None)
.unwrap()
.is_none()
);
config.profile.chat.enabled = true;
assert!(
SubjektivChatFeature::from_resolved_config(&config, None)
.err()
.unwrap()
.to_string()
.contains("explicit live Subject")
);
let host = crate::subjektiv::test_connection(Arc::new(ChatBackendFixture::default()));
let feature = SubjektivChatFeature::from_resolved_config(&config, Some(host.clone()))
.unwrap()
.unwrap();
let descriptor = feature.descriptor();
assert_eq!(
descriptor
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>(),
[READ_TOOL, SEND_TOOL]
);
assert!(
descriptor.hooks.is_empty(),
"no assistant/output/turn hooks may publish chat"
);
config.profile.consolidation_tools = true;
assert!(SubjektivChatFeature::from_resolved_config(&config, Some(host)).is_err());
}
#[tokio::test]
async fn registry_installs_only_explicit_tools_and_does_no_startup_io() {
let client = Arc::new(ChatBackendFixture::default());
let feature = SubjektivChatFeature {
host: crate::subjektiv::test_connection(client.clone()),
};
let mut tools = Vec::new();
let mut hooks = crate::hook::HookRegistryBuilder::new();
let report = crate::feature::FeatureRegistryBuilder::new()
.with_module(feature)
.install_into_pending(&mut tools, &mut hooks);
assert!(!report.has_errors());
assert_eq!(
tools.iter().map(|tool| tool().0.name).collect::<Vec<_>>(),
[READ_TOOL, SEND_TOOL]
);
assert!(client.requests.lock().unwrap().is_empty());
let (_, read) = tools[0]();
let (_, send) = tools[1]();
read.execute(
r#"{"conversation_id":"conversation-1"}"#,
ToolExecutionContext::new("read", "batch", 0),
)
.await
.unwrap();
for ordinal in 0..2 {
let mut input = send_input();
input["operation_id"] = serde_json::json!(format!("explicit-send-{ordinal}"));
send.execute(
&input.to_string(),
ToolExecutionContext::new("send", "batch", ordinal),
)
.await
.unwrap();
}
assert_eq!(
client.chat_requests().len(),
3,
"one read and two explicit sends, no implicit publication"
);
}
#[test]
fn schemas_are_explicit_text_only_and_never_accept_identity() {
let send = schema::<ChatSendMessageRequest>();
let read = schema::<ReadParams>();
for schema in [&send, &read] {
assert_eq!(schema["additionalProperties"], false);
for forbidden in [
"sender",
"subject_id",
"account_id",
"session_id",
"worker_id",
"attachments",
] {
assert!(
schema["properties"].get(forbidden).is_none(),
"{forbidden} must not be model input"
);
}
}
for required in [
"conversation_id",
"operation_id",
"text",
"expected_last_user_message_id",
] {
assert!(
send["required"]
.as_array()
.unwrap()
.contains(&serde_json::json!(required))
);
}
assert!(read["properties"].get("after_sequence").is_some());
}
#[tokio::test]
async fn send_preserves_observed_input_reply_work_and_stable_retry_identity() {
let client = Arc::new(ChatBackendFixture::default());
let tool = tool(client.clone(), Operation::Send);
let input = send_input();
for _ in 0..2 {
let output = invoke(&tool, input.clone()).await.unwrap();
let committed: ChatMessage =
serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
assert_eq!(committed.id, "message-committed");
assert_eq!(committed.operation_id, input["operation_id"]);
assert_eq!(
committed.last_user_message_id.as_deref(),
Some("user-message-3")
);
assert!(output.attachments.is_empty());
}
let requests = client.chat_requests();
assert_eq!(requests.len(), 2);
assert_eq!(
serde_json::to_value(&requests[0]).unwrap(),
serde_json::to_value(&requests[1]).unwrap()
);
let ChatWorkerOperation::Send { request } = &requests[0].operation else {
panic!("Send required")
};
assert_eq!(serde_json::to_value(request).unwrap(), input);
// The fixture confirms forwarding, not storage exactly-once or real model quality.
}
#[tokio::test]
async fn read_preserves_cursor_and_returns_last_user_message_with_small_default_page() {
let client = Arc::new(ChatBackendFixture::default());
let tool = tool(client.clone(), Operation::Read);
let output = invoke(
&tool,
serde_json::json!({"conversation_id":"conversation-1","after_sequence":6}),
)
.await
.unwrap();
let page: ChatMessagePage = serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
assert_eq!(
page.conversation.last_user_message_id.as_deref(),
Some("user-message-3")
);
let requests = client.chat_requests();
let ChatWorkerOperation::Read {
conversation_id,
query,
} = &requests[0].operation
else {
panic!("Read required")
};
assert_eq!(conversation_id, "conversation-1");
assert_eq!(query.after_sequence, Some(6));
assert_eq!(query.limit, Some(5));
assert_eq!(
requests.len(),
1,
"read must not automatically send anything"
);
}
#[tokio::test]
async fn read_and_send_use_distinct_live_bound_subject_permissions() {
for (operation, input, allowed) in [
(
Operation::Read,
serde_json::json!({"conversation_id":"conversation-1"}),
"chat.send",
),
(Operation::Send, send_input(), "chat.read"),
] {
let client = Arc::new(ChatBackendFixture::default());
*client.allowed.lock().unwrap() = vec![allowed];
let tool = tool(client.clone(), operation);
let error = invoke(&tool, input.clone()).await.err().unwrap();
assert!(error.to_string().contains("permission denied"));
assert!(
client.chat_requests().is_empty(),
"denied content must not be read or written"
);
*client.allowed.lock().unwrap() = vec!["chat.read", "chat.send"];
invoke(&tool, input.clone()).await.unwrap();
client.allowed.lock().unwrap().clear();
assert!(
invoke(&tool, input).await.is_err(),
"saved connection must not cache permission"
);
assert_eq!(client.chat_requests().len(), 1);
}
}
#[tokio::test]
async fn backend_rejection_conflict_and_unknown_acceptance_never_report_sent_or_auto_retry() {
for status in [403, 409, 500] {
let client = Arc::new(ChatBackendFixture::default());
client.status.store(status, Ordering::SeqCst);
let error = invoke(&tool(client.clone(), Operation::Send), send_input())
.await
.err()
.unwrap();
if status == 409 {
assert!(matches!(error, ToolError::StructuredConflict { .. }));
} else {
assert!(matches!(error, ToolError::ExecutionFailed(_)));
}
assert_eq!(client.chat_requests().len(), 1);
}
let client = Arc::new(ChatBackendFixture::default());
client.unavailable.store(true, Ordering::SeqCst);
assert!(
invoke(&tool(client.clone(), Operation::Send), send_input())
.await
.err()
.unwrap()
.to_string()
.contains("acceptance unknown")
);
assert_eq!(client.chat_requests().len(), 1);
}
#[tokio::test]
async fn invalid_or_identity_claiming_requests_are_rejected_before_host_access() {
let client = Arc::new(ChatBackendFixture::default());
let send = tool(client.clone(), Operation::Send);
for forbidden in ["sender", "subject_id", "account_id", "attachments"] {
let mut input = send_input();
input[forbidden] = serde_json::json!("forged");
assert!(matches!(
invoke(&send, input).await,
Err(ToolError::InvalidArgument(_))
));
}
for (field, value) in [
("conversation_id", serde_json::json!(" ")),
("operation_id", serde_json::json!("")),
("text", serde_json::json!(" ")),
(
"text",
serde_json::json!("x".repeat(server_api::CHAT_MAX_TEXT_BYTES + 1)),
),
("expected_last_user_message_id", serde_json::json!(-1)),
] {
let mut input = send_input();
input[field] = value;
assert!(matches!(
invoke(&send, input).await,
Err(ToolError::InvalidArgument(_))
));
}
let read = tool(client.clone(), Operation::Read);
for limit in [0, server_api::CHAT_MAX_PAGE_LIMIT + 1] {
assert!(
invoke(
&read,
serde_json::json!({"conversation_id":"conversation-1","limit":limit})
)
.await
.is_err()
);
}
assert!(client.requests.lock().unwrap().is_empty());
}
#[tokio::test]
async fn unexpected_response_variants_and_oversized_reads_fail_closed() {
let client = Arc::new(ChatBackendFixture::default());
client.wrong_variant.store(true, Ordering::SeqCst);
assert!(
invoke(&tool(client.clone(), Operation::Send), send_input())
.await
.err()
.unwrap()
.to_string()
.contains("unexpected chat")
);
client
.page_text_bytes
.store(MAX_TOOL_OUTPUT_BYTES + 1, Ordering::SeqCst);
let error = invoke(
&tool(client.clone(), Operation::Read),
serde_json::json!({"conversation_id":"conversation-1"}),
)
.await
.err()
.unwrap();
assert!(error.to_string().contains("smaller page"));
assert_eq!(client.chat_requests().len(), 2);
}
crates/worker/src/permission.rs
+48
View File
@@ -52,6 +52,28 @@ impl Hook<PreToolCall> for PermissionHook {
}
}
// Only prune schemas when every possible argument is denied. Conditional
// rules remain visible and are checked by the normal execution boundary.
// This is surface composition, not another permission source.
pub(crate) fn tool_is_unconditionally_denied(
config: &ToolPermissionConfig,
tool_name: &str,
) -> bool {
for rule in config
.rules
.iter()
.filter(|rule| rule.tool.eq_ignore_ascii_case(tool_name))
{
if rule.action != ToolPermissionAction::Deny {
return false;
}
if !rule.pattern.is_empty() && rule.pattern.bytes().all(|byte| byte == b'*') {
return true;
}
}
config.default_action == ToolPermissionAction::Deny
}
pub(crate) fn permission_action_for(
config: &ToolPermissionConfig,
tool_name: &str,
@@ -138,6 +160,32 @@ mod tests {
}
}
#[test]
fn static_tool_pruning_respects_rule_order_and_never_hides_conditional_allow() {
let mut policy = ToolPermissionConfig {
default_action: ToolPermissionAction::Deny,
rules: Vec::new(),
};
assert!(tool_is_unconditionally_denied(&policy, "Bash"));
policy.rules.push(ToolPermissionRule {
tool: "bash".into(),
pattern: "git *".into(),
action: ToolPermissionAction::Allow,
});
assert!(!tool_is_unconditionally_denied(&policy, "Bash"));
policy.rules.push(ToolPermissionRule {
tool: "Bash".into(),
pattern: "*".into(),
action: ToolPermissionAction::Deny,
});
assert!(!tool_is_unconditionally_denied(&policy, "Bash"));
policy.rules.swap(0, 1);
assert!(tool_is_unconditionally_denied(&policy, "Bash"));
policy.default_action = ToolPermissionAction::Allow;
assert!(tool_is_unconditionally_denied(&policy, "Bash"));
assert!(!tool_is_unconditionally_denied(&policy, "Read"));
}
#[test]
fn first_matching_rule_wins_by_declaration_order() {
let hook = PermissionHook::new(ToolPermissionConfig {
crates/worker/src/spawn/tool.rs
+1
View File
@@ -272,6 +272,7 @@ impl ParentNotificationTarget {
pub(crate) fn notify(&self, child_session_id: String, message: String) {
let method = Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: protocol::new_submission_request_id(),
message,
source: protocol::AuthenticatedInputSource::SubWorker {
crates/worker/src/subjektiv.rs
+59
View File
@@ -162,6 +162,17 @@ pub trait SubjektivHost: Send + Sync + 'static {
fn workspace_id(&self) -> Option<String> {
None
}
/// Text-only public conversation access. Restricted Job and standalone Hosts
/// inherit denial, not the connected Subject body's posting capability.
fn chat(
&self,
_context: &SubjektivHostContext,
_operation: ChatWorkerOperation,
) -> Result<ChatWorkerResponse, SubjektivHostError> {
Err(SubjektivHostError::PermissionDenied(
"conversation access is not granted by this Host".into(),
))
}
fn memory(
&self,
context: &SubjektivHostContext,
@@ -399,6 +410,18 @@ impl SubjektivHost for BackendSubjektivHost {
fn consolidation_granted(&self) -> bool {
self.consolidation_granted
}
fn chat(
&self,
context: &SubjektivHostContext,
operation: ChatWorkerOperation,
) -> Result<ChatWorkerResponse, SubjektivHostError> {
self.authorize_operation(context, chat_operation_name(&operation))?;
// Backend revalidates current singleton ownership and conversation access
// at the read/write boundary; this preflight cannot replace those checks.
self.execute(WorkspaceServerOperation::ChatWorker(ChatWorkerRequest {
operation,
}))
}
fn memory(
&self,
context: &SubjektivHostContext,
@@ -574,6 +597,13 @@ impl JournalJobHost for BackendJournalJobHost {
}
}
fn chat_operation_name(operation: &ChatWorkerOperation) -> &'static str {
match operation {
ChatWorkerOperation::Read { .. } => "chat.read",
ChatWorkerOperation::Send { .. } => "chat.send",
}
}
/// Live connection with Worker-supplied authority. Not serializable and never
/// inherited by an ordinary SubWorker.
#[derive(Clone)]
@@ -632,6 +662,13 @@ impl SubjektivHostConnection {
) -> Result<MemoryJournalResponse, SubjektivHostError> {
self.host.memory_journal(&self.context, request)
}
pub fn chat(
&self,
operation: ChatWorkerOperation,
) -> Result<ChatWorkerResponse, SubjektivHostError> {
self.authorize_operation(chat_operation_name(&operation))?;
self.host.chat(&self.context, operation)
}
pub fn memory(
&self,
operation: SubjektivMemoryBackendOperation,
@@ -769,6 +806,28 @@ mod tests {
use std::pin::Pin;
use std::sync::Mutex;
#[test]
fn non_chat_local_hosts_do_not_inherit_public_conversation_access() {
let host = Arc::new(LocalHost::default());
let context = SubjektivHostContext {
worker_id: "worker".into(),
session_id: "session".into(),
};
let read = ChatWorkerOperation::Read {
conversation_id: "conversation".into(),
query: ChatMessagePageQuery::default(),
};
assert!(matches!(
host.chat(&context, read.clone()),
Err(SubjektivHostError::PermissionDenied(_))
));
let connection = SubjektivHostConnection::new(host, context).unwrap();
assert!(matches!(
connection.chat(read),
Err(SubjektivHostError::PermissionDenied(_))
));
}
#[derive(Default)]
struct LocalHost {
startup: Option<JournalStartupSnapshot>,
crates/worker/src/worker.rs
+123
View File
@@ -792,6 +792,7 @@ pub enum WorkspaceServerOperation {
WorkerObservationCapture(server_api::WorkerObservationSubjectRef),
SubjektivStageCandidate(server_api::SubjektivStageCandidateRequest),
SubjektivRecordSession(server_api::SubjektivRecordSessionRequest),
ChatWorker(server_api::ChatWorkerRequest),
SubjektivMemory(server_api::SubjektivMemoryBackendRequest),
SubjektivSession(server_api::SubjektivSessionBackendRequest),
}
@@ -1025,6 +1026,9 @@ fn workspace_server_operation_request(
WorkspaceServerOperation::SubjektivRecordSession(request) => {
workspace_server_json_request(format!("{base}/subjektiv/sessions"), &request)
}
WorkspaceServerOperation::ChatWorker(request) => {
workspace_server_json_request(format!("{base}/subjektiv/chat/worker"), &request)
}
WorkspaceServerOperation::SubjektivMemory(request) => {
workspace_server_json_request(format!("{base}/subjektiv/memory"), &request)
}
@@ -12940,6 +12944,19 @@ mod build_summary_prompt_tests {
{
1
}
LogEntry::Extension {
domain, payload, ..
} if domain == SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN
&& payload["pending_notifications"]
.as_array()
.is_some_and(|messages| {
messages
.iter()
.any(|message| message["message"] == "persist-failure")
}) =>
{
6
}
LogEntry::AnnotatedSystemItem { entry, .. }
if input_schema(entry) == Some("feature:invocation-result") =>
{
@@ -18319,6 +18336,112 @@ permission = "write"
assert_eq!(committed.annotation.origin, provenance);
}
#[test]
fn notification_admission_save_failure_leaves_no_receipt_and_same_request_can_retry() {
let temp = tempfile::tempdir().unwrap();
let original = PendingSubmissionHandle::for_test(temp.path());
let handle = PendingSubmissionHandle {
state: original.state.clone(),
writer: LogWriterHandle {
store: FailRunTerminalStore {
inner: original.writer.store.clone(),
fail_run_terminal: Arc::new(AtomicBool::new(false)),
invocation_failure: Arc::new(AtomicUsize::new(6)),
},
state: original.writer.state.clone(),
sink: original.writer.sink.clone(),
in_flight: None,
},
};
assert!(
handle
.accept_notification("stable-request".into(), "persist-failure".into())
.is_err()
);
assert_eq!(handle.snapshot().notification_count, 0);
assert!(original.persisted_entries_for_test().is_empty());
assert!(
handle
.state
.lock()
.unwrap()
.notification_receipts
.is_empty()
);
assert!(
handle
.accept_notification("stable-request".into(), "persist-failure".into())
.unwrap()
);
assert!(
!handle
.accept_notification("stable-request".into(), "persist-failure".into())
.unwrap()
);
let saved = original.persisted_entries_for_test();
let LogEntry::Extension { payload, .. } = saved.last().unwrap() else {
panic!("saved admission");
};
let state: PendingActivationState = serde_json::from_value(payload.clone()).unwrap();
assert_eq!(state.pending_notifications.len(), 1);
assert_eq!(state.notification_receipts.len(), 1);
}
#[test]
fn notification_admission_limits_do_not_persist_rejection_or_evict_accepted_input() {
let temp = tempfile::tempdir().unwrap();
let handle = PendingSubmissionHandle::for_test(temp.path());
for index in 0..MAX_PENDING_SUBMISSIONS {
assert!(
handle
.accept_notification(format!("notice-{index}"), "bounded".into())
.unwrap()
);
}
let saved = handle.persisted_entries_for_test();
assert!(matches!(
handle.accept_notification("overflow".into(), "bounded".into()),
Err(PendingSubmissionError::CountLimit)
));
assert_eq!(handle.persisted_entries_for_test().len(), saved.len());
assert_eq!(
handle.snapshot().notification_count as usize,
MAX_PENDING_SUBMISSIONS
);
// An exact retry remains accepted even at capacity; a changed body never does.
assert!(
!handle
.accept_notification("notice-0".into(), "bounded".into())
.unwrap()
);
assert!(matches!(
handle.accept_notification("notice-0".into(), "changed".into()),
Err(PendingSubmissionError::IdempotencyConflict)
));
let bytes = tempfile::tempdir().unwrap();
let handle = PendingSubmissionHandle::for_test(bytes.path());
assert!(
handle
.accept_notification(
"full-byte-budget".into(),
"x".repeat(MAX_PENDING_SUBMISSION_BYTES as usize)
)
.unwrap()
);
let saved = handle.persisted_entries_for_test();
assert!(matches!(
handle.accept_notification("byte-overflow".into(), "x".into()),
Err(PendingSubmissionError::ByteLimit)
));
assert_eq!(handle.persisted_entries_for_test().len(), saved.len());
let LogEntry::Extension { payload, .. } = saved.last().unwrap() else {
panic!("saved admission");
};
let state: PendingActivationState = serde_json::from_value(payload.clone()).unwrap();
assert_eq!(state.pending_notifications.len(), 1);
assert_eq!(state.notification_receipts.len(), 1);
}
#[test]
fn notification_retry_identity_is_scoped_to_authenticated_source() {
let temp = tempfile::tempdir().unwrap();
crates/worker/src/worker_composition_tests.rs
+66
View File
@@ -67,6 +67,72 @@ fn assert_publication(
assert_eq!(prompt.contains("## SubWorker orchestration"), sub_worker);
}
#[tokio::test]
async fn async_chat_profile_prunes_execution_tools_without_granting_authority() {
let temp = tempfile::tempdir().unwrap();
let _sandbox = worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
let resolved = manifest::ProfileResolver::new()
.with_workspace_base(temp.path())
.resolve(
&manifest::ProfileSelector::source_named(
manifest::ProfileRegistrySource::Builtin,
"async-chat",
),
manifest::ProfileResolveOptions::with_worker_name("chat-composition"),
)
.unwrap();
let mut manifest = resolved.manifest;
// This case isolates registration restrictions from independently tested
// Subject startup Memory/Journal authorization and chat Host requirements.
manifest.feature.subjektiv.profile.enabled = false;
let (_, store) = feature_restore_fixture(&temp);
let client = Arc::new(FeatureAdmissionClient::default());
let controller = bootstrap(&temp, manifest, store, Some(client.clone()))
.prepare()
.await
.unwrap()
.start()
.await
.unwrap();
let names = controller.handle.shared_state.greeting().tools;
for forbidden in [
"Read",
"Write",
"Edit",
"Glob",
"Grep",
"Bash",
"SubWorkerSpawn",
"WebFetch",
"WebSearch",
"SpawnTicketWorker",
"JobRun",
] {
assert!(
!names.iter().any(|name| name == forbidden),
"chat body exposed {forbidden}: {names:?}"
);
}
for required in [
"WorkerSpawn",
"WorkerNotify",
"WorkerStop",
"WorkerList",
"QueryObjective",
"ShowTicket",
] {
assert!(
names.iter().any(|name| name == required),
"chat body lacks {required}: {names:?}"
);
}
assert!(
client.requests.lock().unwrap().is_empty(),
"Profile activation must not issue authority or read protected content"
);
stop_body_publication_controller(controller).await;
}
#[tokio::test]
async fn objective_and_sub_worker_saved_composition_survives_create_and_restore_matrix() {
for mode in [manifest::WorkerMode::Tools, manifest::WorkerMode::Wip] {
crates/worker/tests/compact_events_test.rs
+1
View File
@@ -1935,6 +1935,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() {
let completion_request_id = "subworker-completion-during-compact";
handle
.send(Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: completion_request_id.into(),
message: "subworker completed".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire,
crates/worker/tests/feature_invocation_test.rs
+1
View File
@@ -601,6 +601,7 @@ fn notify_rejects_typed_side_effect_message_payloads() {
message: "/prepare(first)".into(),
},
protocol::Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "n-1".into(),
message: "/prepare(first)".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire,