merge: integrate ticket assignment notifications
# Conflicts: # crates/workspace-server/src/store.rs
This commit is contained in:
@@ -9,6 +9,7 @@ use std::fmt;
|
||||
use std::fs::{self, File, OpenOptions};
|
||||
use std::io::{self, Write};
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use chrono::Utc;
|
||||
use fs4::fs_std::FileExt;
|
||||
@@ -2274,11 +2275,40 @@ impl LocalTicketBackend {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct SqliteTicketMutationEvent {
|
||||
pub workspace_id: String,
|
||||
pub ticket_id: String,
|
||||
pub event_index: i64,
|
||||
pub event_kind: TicketEventKind,
|
||||
}
|
||||
|
||||
pub type SqliteTicketMutationHook =
|
||||
dyn Fn(&Connection, &SqliteTicketMutationEvent) -> Result<()> + Send + Sync;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SqliteTicketBackend {
|
||||
db_path: PathBuf,
|
||||
workspace_id: String,
|
||||
record_language: Option<String>,
|
||||
event_attributes: BTreeMap<String, String>,
|
||||
mutation_hook: Option<Arc<SqliteTicketMutationHook>>,
|
||||
}
|
||||
|
||||
impl fmt::Debug for SqliteTicketBackend {
|
||||
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
formatter
|
||||
.debug_struct("SqliteTicketBackend")
|
||||
.field("db_path", &self.db_path)
|
||||
.field("workspace_id", &self.workspace_id)
|
||||
.field("record_language", &self.record_language)
|
||||
.field("event_attributes", &self.event_attributes)
|
||||
.field(
|
||||
"mutation_hook",
|
||||
&self.mutation_hook.as_ref().map(|_| "configured"),
|
||||
)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl SqliteTicketBackend {
|
||||
@@ -2287,9 +2317,21 @@ impl SqliteTicketBackend {
|
||||
db_path: db_path.into(),
|
||||
workspace_id: workspace_id.into(),
|
||||
record_language: None,
|
||||
event_attributes: BTreeMap::new(),
|
||||
mutation_hook: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_event_attributes(mut self, attributes: BTreeMap<String, String>) -> Self {
|
||||
self.event_attributes = attributes;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_mutation_hook(mut self, hook: Arc<SqliteTicketMutationHook>) -> Self {
|
||||
self.mutation_hook = Some(hook);
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_record_language(mut self, language: Option<&str>) -> Self {
|
||||
self.record_language = language.and_then(normalized_record_language);
|
||||
self
|
||||
@@ -2506,10 +2548,25 @@ CREATE TABLE IF NOT EXISTS typed_ticket_artifacts (
|
||||
conn.execute("INSERT INTO typed_ticket_event_references (workspace_id, ticket_id, event_index, ordinal, kind, target) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
||||
params![self.workspace_id, ticket_id, next_index, ordinal as i64, reference.kind, reference.target]).map_err(sqlite_err)?;
|
||||
}
|
||||
for (key, value) in &event.attributes {
|
||||
let mut attributes = event.attributes.clone();
|
||||
for (key, value) in &self.event_attributes {
|
||||
attributes.insert(key.clone(), value.clone());
|
||||
}
|
||||
for (key, value) in &attributes {
|
||||
conn.execute("INSERT INTO typed_ticket_event_attributes (workspace_id, ticket_id, event_index, key, value) VALUES (?1, ?2, ?3, ?4, ?5)",
|
||||
params![self.workspace_id, ticket_id, next_index, key, value]).map_err(sqlite_err)?;
|
||||
}
|
||||
if let Some(hook) = &self.mutation_hook {
|
||||
hook(
|
||||
conn,
|
||||
&SqliteTicketMutationEvent {
|
||||
workspace_id: self.workspace_id.clone(),
|
||||
ticket_id: ticket_id.to_string(),
|
||||
event_index: next_index,
|
||||
event_kind: event.kind.clone(),
|
||||
},
|
||||
)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -2718,6 +2775,9 @@ CREATE TABLE IF NOT EXISTS typed_ticket_artifacts (
|
||||
for row in rows {
|
||||
let (index, kind, author, at, status, from, to, reason, state_field, heading, body) =
|
||||
row.map_err(sqlite_err)?;
|
||||
let mut attributes = self.load_event_attributes(conn, ticket_id, index)?;
|
||||
attributes.insert("event_id".to_string(), format!("{ticket_id}:{index}"));
|
||||
attributes.insert("event_sequence".to_string(), index.to_string());
|
||||
events.push(TicketEvent {
|
||||
kind: TicketEventKind::from(kind.as_str()),
|
||||
author,
|
||||
@@ -2730,7 +2790,7 @@ CREATE TABLE IF NOT EXISTS typed_ticket_artifacts (
|
||||
heading,
|
||||
body: MarkdownText::new(body),
|
||||
references: self.load_event_references(conn, ticket_id, index)?,
|
||||
attributes: self.load_event_attributes(conn, ticket_id, index)?,
|
||||
attributes,
|
||||
});
|
||||
}
|
||||
Ok(events)
|
||||
@@ -6393,6 +6453,41 @@ state: planning
|
||||
assert_partial_body_replacement_semantics(&backend);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sqlite_mutation_hook_failure_rolls_back_ticket_event() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let db_path = tmp.path().join("workspace.db");
|
||||
let backend = SqliteTicketBackend::new(&db_path, "workspace-test");
|
||||
let created = backend.create(NewTicket::new("Atomic mutation")).unwrap();
|
||||
let before = backend
|
||||
.show(TicketIdOrSlug::Id(created.id.clone()))
|
||||
.unwrap()
|
||||
.events
|
||||
.len();
|
||||
let failing = backend.clone().with_mutation_hook(Arc::new(|_, event| {
|
||||
Err(TicketError::Conflict(format!(
|
||||
"reject outbox for {}:{}",
|
||||
event.ticket_id, event.event_index
|
||||
)))
|
||||
}));
|
||||
assert!(
|
||||
failing
|
||||
.add_event(
|
||||
TicketIdOrSlug::Id(created.id.clone()),
|
||||
NewTicketEvent::new(TicketEventKind::Comment, "must roll back"),
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
let after = backend.show(TicketIdOrSlug::Id(created.id)).unwrap();
|
||||
assert_eq!(after.events.len(), before);
|
||||
assert!(
|
||||
after
|
||||
.events
|
||||
.iter()
|
||||
.all(|event| event.body.as_str() != "must roll back")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sqlite_backend_persists_core_ticket_operations() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
|
||||
+148
-79
@@ -34,12 +34,15 @@ const MAX_BODY_MAX_BYTES: usize = 64 * 1024;
|
||||
const DEFAULT_DIAGNOSTIC_LIMIT: usize = 100;
|
||||
const MAX_DIAGNOSTIC_LIMIT: usize = 500;
|
||||
|
||||
pub const TICKET_BASE_TOOL_NAMES: [&str; 12] = [
|
||||
pub const TICKET_BASE_TOOL_NAMES: [&str; 15] = [
|
||||
"TicketCreate",
|
||||
"TicketEditItem",
|
||||
"TicketList",
|
||||
"TicketShow",
|
||||
"TicketComment",
|
||||
"TicketPlan",
|
||||
"TicketDecision",
|
||||
"TicketImplementationReport",
|
||||
"TicketReview",
|
||||
"TicketIntakeReady",
|
||||
"TicketQueue",
|
||||
@@ -66,12 +69,15 @@ pub const TICKET_ORCHESTRATION_TOOL_NAMES: [&str; 4] = [
|
||||
pub const TICKET_ORCHESTRATION_READ_ONLY_TOOL_NAMES: [&str; 2] =
|
||||
["TicketRelationQuery", "TicketOrchestrationPlanQuery"];
|
||||
|
||||
pub const TICKET_TOOL_NAMES: [&str; 16] = [
|
||||
pub const TICKET_TOOL_NAMES: [&str; 19] = [
|
||||
"TicketCreate",
|
||||
"TicketEditItem",
|
||||
"TicketList",
|
||||
"TicketShow",
|
||||
"TicketComment",
|
||||
"TicketPlan",
|
||||
"TicketDecision",
|
||||
"TicketImplementationReport",
|
||||
"TicketReview",
|
||||
"TicketIntakeReady",
|
||||
"TicketQueue",
|
||||
@@ -94,10 +100,13 @@ pub const TICKET_READ_ONLY_TOOL_NAMES: [&str; 6] = [
|
||||
"TicketOrchestrationPlanQuery",
|
||||
];
|
||||
|
||||
pub const TICKET_MUTATING_TOOL_NAMES: [&str; 10] = [
|
||||
pub const TICKET_MUTATING_TOOL_NAMES: [&str; 13] = [
|
||||
"TicketCreate",
|
||||
"TicketEditItem",
|
||||
"TicketComment",
|
||||
"TicketPlan",
|
||||
"TicketDecision",
|
||||
"TicketImplementationReport",
|
||||
"TicketReview",
|
||||
"TicketIntakeReady",
|
||||
"TicketQueue",
|
||||
@@ -120,9 +129,11 @@ routing, closing, planning, or implementation decisions.";
|
||||
const SHOW_DESCRIPTION: &str = "Show one Ticket by id or exact query through the configured \
|
||||
typed Ticket backend. Output includes bounded Markdown body, recent thread events, resolution, and \
|
||||
artifact metadata.";
|
||||
const COMMENT_DESCRIPTION: &str = "Append a typed Ticket thread event. `role` must be `comment`, \
|
||||
`plan`, `decision`, or `implementation_report`; `body` is Markdown. Writes stay inside the \
|
||||
configured Ticket backend root.";
|
||||
const COMMENT_DESCRIPTION: &str = "Append a typed Ticket comment event. `body` is Markdown.";
|
||||
const PLAN_DESCRIPTION: &str = "Append a typed Ticket plan event. `body` is Markdown.";
|
||||
const DECISION_DESCRIPTION: &str = "Append a typed Ticket decision event. `body` is Markdown.";
|
||||
const IMPLEMENTATION_REPORT_DESCRIPTION: &str =
|
||||
"Append a typed Ticket implementation_report event. `body` is Markdown.";
|
||||
const REVIEW_DESCRIPTION: &str = "Append a Ticket review event. `result` must be `approve` or \
|
||||
`request_changes`; `body` is Markdown. Writes stay inside the configured Ticket backend root.";
|
||||
const INTAKE_READY_DESCRIPTION: &str = "Mark an existing Ticket planning lane ready through the typed \
|
||||
@@ -161,6 +172,9 @@ fn base_tool_description(name: &str) -> &'static str {
|
||||
"TicketList" => LIST_DESCRIPTION,
|
||||
"TicketShow" => SHOW_DESCRIPTION,
|
||||
"TicketComment" => COMMENT_DESCRIPTION,
|
||||
"TicketPlan" => PLAN_DESCRIPTION,
|
||||
"TicketDecision" => DECISION_DESCRIPTION,
|
||||
"TicketImplementationReport" => IMPLEMENTATION_REPORT_DESCRIPTION,
|
||||
"TicketReview" => REVIEW_DESCRIPTION,
|
||||
"TicketIntakeReady" => INTAKE_READY_DESCRIPTION,
|
||||
"TicketQueue" => QUEUE_DESCRIPTION,
|
||||
@@ -361,9 +375,6 @@ struct TicketCreateParams {
|
||||
/// Markdown body for item.md. If omitted, a small default body is used.
|
||||
#[serde(default)]
|
||||
body: Option<String>,
|
||||
/// Optional thread author for the create event.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
/// Optional assignee frontmatter value.
|
||||
#[serde(default)]
|
||||
assignee: Option<String>,
|
||||
@@ -376,9 +387,6 @@ struct TicketCreateParams {
|
||||
/// Optional state frontmatter value. Defaults to `planning`.
|
||||
#[serde(default)]
|
||||
state: Option<TicketWorkflowStateParam>,
|
||||
/// Optional queued_by frontmatter value.
|
||||
#[serde(default)]
|
||||
queued_by: Option<String>,
|
||||
/// Optional queued_at frontmatter value.
|
||||
#[serde(default)]
|
||||
queued_at: Option<String>,
|
||||
@@ -412,9 +420,6 @@ struct TicketEditItemParams {
|
||||
/// Optional target repository/ref update.
|
||||
#[serde(default)]
|
||||
target: Option<crate::TicketTargetEdit>,
|
||||
/// Optional thread author for the audited item_edit event.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, Deserialize, schemars::JsonSchema)]
|
||||
@@ -542,25 +547,11 @@ struct TicketShowParams {
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum TicketCommentRoleParam {
|
||||
Comment,
|
||||
Plan,
|
||||
Decision,
|
||||
ImplementationReport,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
struct TicketCommentParams {
|
||||
struct TicketThreadEventParams {
|
||||
/// Ticket id.
|
||||
ticket: String,
|
||||
/// Thread event role: `comment`, `plan`, `decision`, or `implementation_report`.
|
||||
role: TicketCommentRoleParam,
|
||||
/// Markdown event body.
|
||||
body: String,
|
||||
/// Optional thread author.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
@@ -578,9 +569,6 @@ struct TicketReviewParams {
|
||||
result: TicketReviewResultParam,
|
||||
/// Markdown review body.
|
||||
body: String,
|
||||
/// Optional thread author.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
@@ -589,9 +577,6 @@ struct TicketIntakeReadyParams {
|
||||
ticket: String,
|
||||
/// Concise bounded intake summary to append as a typed intake_summary event.
|
||||
intake_summary: String,
|
||||
/// Optional author for both intake_summary and state_changed events.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
/// Reason attached to the state_changed event. Defaults to `planning_ready`.
|
||||
#[serde(default)]
|
||||
reason: Option<String>,
|
||||
@@ -604,9 +589,6 @@ struct TicketIntakeReadyParams {
|
||||
struct TicketQueueParams {
|
||||
/// Ticket id.
|
||||
ticket: String,
|
||||
/// Optional queued_by frontmatter value. Defaults to the backend/user default.
|
||||
#[serde(default)]
|
||||
queued_by: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
@@ -621,9 +603,6 @@ struct TicketWorkflowStateParams {
|
||||
reason: String,
|
||||
/// Markdown body for the typed state_changed event.
|
||||
body: String,
|
||||
/// Optional thread author.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
@@ -673,9 +652,6 @@ struct TicketRelationRecordParams {
|
||||
/// Optional bounded rationale/note.
|
||||
#[serde(default)]
|
||||
note: Option<String>,
|
||||
/// Optional record author.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
@@ -757,9 +733,6 @@ struct TicketOrchestrationPlanRecordParams {
|
||||
/// Accepted plan fields. Required for accepted_plan and invalid for other kinds.
|
||||
#[serde(default)]
|
||||
accepted_plan: Option<AcceptedOrchestrationPlanParams>,
|
||||
/// Optional record author.
|
||||
#[serde(default)]
|
||||
author: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
@@ -851,6 +824,21 @@ struct TicketCommentTool {
|
||||
backend: TicketToolBackend,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct TicketPlanTool {
|
||||
backend: TicketToolBackend,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct TicketDecisionTool {
|
||||
backend: TicketToolBackend,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct TicketImplementationReportTool {
|
||||
backend: TicketToolBackend,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct TicketReviewTool {
|
||||
backend: TicketToolBackend,
|
||||
@@ -918,12 +906,12 @@ impl Tool for TicketCreateTool {
|
||||
if let Some(body) = params.body {
|
||||
input.body = MarkdownText::new(body);
|
||||
}
|
||||
input.author = params.author;
|
||||
input.author = None;
|
||||
input.assignee = params.assignee;
|
||||
input.readiness = params.readiness;
|
||||
input.risk_flags = params.risk_flags;
|
||||
input.workflow_state = params.state.map(TicketWorkflowStateParam::into_state);
|
||||
input.queued_by = params.queued_by;
|
||||
input.queued_by = None;
|
||||
input.queued_at = params.queued_at;
|
||||
input.repository_id = params.repository_id;
|
||||
input.ref_selector = params.ref_selector;
|
||||
@@ -971,7 +959,7 @@ impl Tool for TicketEditItemTool {
|
||||
body: params.body.map(MarkdownText::new),
|
||||
body_replacement,
|
||||
target: params.target,
|
||||
author: params.author,
|
||||
author: None,
|
||||
};
|
||||
let ticket = self
|
||||
.backend
|
||||
@@ -1066,6 +1054,26 @@ impl Tool for TicketShowTool {
|
||||
}
|
||||
}
|
||||
|
||||
fn execute_ticket_thread_event(
|
||||
backend: &TicketToolBackend,
|
||||
tool_name: &str,
|
||||
kind: TicketEventKind,
|
||||
input_json: &str,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let params: TicketThreadEventParams = parse_input(tool_name, input_json)?;
|
||||
let role = kind.as_str().to_string();
|
||||
backend
|
||||
.add_event(
|
||||
TicketIdOrSlug::Query(params.ticket.clone()),
|
||||
NewTicketEvent::new(kind, params.body),
|
||||
)
|
||||
.map_err(|error| backend_error(tool_name, error))?;
|
||||
Ok(json_output(
|
||||
format!("Appended {role} event to ticket {}", params.ticket),
|
||||
json!({ "ticket": params.ticket, "event": role, "ok": true }),
|
||||
))
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for TicketCommentTool {
|
||||
async fn execute(
|
||||
@@ -1073,26 +1081,42 @@ impl Tool for TicketCommentTool {
|
||||
input_json: &str,
|
||||
_ctx: llm_engine::tool::ToolExecutionContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let params: TicketCommentParams = parse_input("TicketComment", input_json)?;
|
||||
let kind = match params.role {
|
||||
TicketCommentRoleParam::Comment => TicketEventKind::Comment,
|
||||
TicketCommentRoleParam::Plan => TicketEventKind::Plan,
|
||||
TicketCommentRoleParam::Decision => TicketEventKind::Decision,
|
||||
TicketCommentRoleParam::ImplementationReport => TicketEventKind::ImplementationReport,
|
||||
};
|
||||
let role = kind.as_str().to_string();
|
||||
let mut event = NewTicketEvent::new(kind, params.body);
|
||||
event.author = params.author;
|
||||
self.backend
|
||||
.add_event(TicketIdOrSlug::Query(params.ticket.clone()), event)
|
||||
.map_err(|error| backend_error("TicketComment", error))?;
|
||||
Ok(json_output(
|
||||
format!("Appended {role} event to ticket {}", params.ticket),
|
||||
json!({ "ticket": params.ticket, "event": role, "ok": true }),
|
||||
))
|
||||
execute_ticket_thread_event(
|
||||
&self.backend,
|
||||
"TicketComment",
|
||||
TicketEventKind::Comment,
|
||||
input_json,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! impl_ticket_thread_event_tool {
|
||||
($tool:ty, $name:literal, $kind:expr) => {
|
||||
#[async_trait]
|
||||
impl Tool for $tool {
|
||||
async fn execute(
|
||||
&self,
|
||||
input_json: &str,
|
||||
_ctx: llm_engine::tool::ToolExecutionContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
execute_ticket_thread_event(&self.backend, $name, $kind, input_json)
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
impl_ticket_thread_event_tool!(TicketPlanTool, "TicketPlan", TicketEventKind::Plan);
|
||||
impl_ticket_thread_event_tool!(
|
||||
TicketDecisionTool,
|
||||
"TicketDecision",
|
||||
TicketEventKind::Decision
|
||||
);
|
||||
impl_ticket_thread_event_tool!(
|
||||
TicketImplementationReportTool,
|
||||
"TicketImplementationReport",
|
||||
TicketEventKind::ImplementationReport
|
||||
);
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for TicketReviewTool {
|
||||
async fn execute(
|
||||
@@ -1108,7 +1132,7 @@ impl Tool for TicketReviewTool {
|
||||
let result_str = result.as_str().to_string();
|
||||
let review = TicketReview {
|
||||
result,
|
||||
author: params.author,
|
||||
author: None,
|
||||
body: MarkdownText::new(params.body),
|
||||
};
|
||||
self.backend
|
||||
@@ -1138,14 +1162,14 @@ impl Tool for TicketIntakeReadyTool {
|
||||
.default_intake_ready_state_change_body(from.as_str())
|
||||
});
|
||||
let mut summary = TicketIntakeSummary::new(params.intake_summary);
|
||||
summary.author = params.author.clone();
|
||||
summary.author = None;
|
||||
let mut change = TicketStateChange::new(
|
||||
from.as_str(),
|
||||
TicketWorkflowState::Ready.as_str(),
|
||||
reason,
|
||||
body,
|
||||
);
|
||||
change.author = params.author;
|
||||
change.author = None;
|
||||
self.backend
|
||||
.mark_intake_ready(
|
||||
TicketIdOrSlug::Query(params.ticket.clone()),
|
||||
@@ -1168,7 +1192,7 @@ impl Tool for TicketQueueTool {
|
||||
_ctx: llm_engine::tool::ToolExecutionContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let params: TicketQueueParams = parse_input("TicketQueue", input_json)?;
|
||||
let queued_by = params.queued_by.unwrap_or_else(default_author);
|
||||
let queued_by = default_author();
|
||||
self.backend
|
||||
.queue_ready(TicketIdOrSlug::Query(params.ticket.clone()), &queued_by)
|
||||
.map_err(|error| backend_error("TicketQueue", error))?;
|
||||
@@ -1196,7 +1220,7 @@ impl Tool for TicketWorkflowStateTool {
|
||||
}
|
||||
let mut change =
|
||||
TicketStateChange::new(from.as_str(), to.as_str(), params.reason, params.body);
|
||||
change.author = params.author;
|
||||
change.author = None;
|
||||
self.backend
|
||||
.set_workflow_state(TicketIdOrSlug::Query(params.ticket.clone()), change)
|
||||
.map_err(|error| backend_error("TicketWorkflowState", error))?;
|
||||
@@ -1251,7 +1275,7 @@ impl Tool for TicketRelationRecordTool {
|
||||
kind: params.kind.into_kind(),
|
||||
target: params.target.clone(),
|
||||
note: params.note,
|
||||
author: params.author,
|
||||
author: None,
|
||||
};
|
||||
let output = self
|
||||
.backend
|
||||
@@ -1325,7 +1349,7 @@ impl Tool for TicketOrchestrationPlanRecordTool {
|
||||
related_ticket: params.related_ticket,
|
||||
note: params.note,
|
||||
accepted_plan,
|
||||
author: params.author,
|
||||
author: None,
|
||||
};
|
||||
let output = self
|
||||
.backend
|
||||
@@ -1703,7 +1727,9 @@ fn input_schema(name: &str) -> Value {
|
||||
"TicketEditItem" => serde_json::to_value(schemars::schema_for!(TicketEditItemParams)),
|
||||
"TicketList" => serde_json::to_value(schemars::schema_for!(TicketListParams)),
|
||||
"TicketShow" => serde_json::to_value(schemars::schema_for!(TicketShowParams)),
|
||||
"TicketComment" => serde_json::to_value(schemars::schema_for!(TicketCommentParams)),
|
||||
"TicketComment" | "TicketPlan" | "TicketDecision" | "TicketImplementationReport" => {
|
||||
serde_json::to_value(schemars::schema_for!(TicketThreadEventParams))
|
||||
}
|
||||
"TicketReview" => serde_json::to_value(schemars::schema_for!(TicketReviewParams)),
|
||||
"TicketIntakeReady" => serde_json::to_value(schemars::schema_for!(TicketIntakeReadyParams)),
|
||||
"TicketQueue" => serde_json::to_value(schemars::schema_for!(TicketQueueParams)),
|
||||
@@ -1747,6 +1773,9 @@ impl_from_backend!(TicketEditItemTool);
|
||||
impl_from_backend!(TicketListTool);
|
||||
impl_from_backend!(TicketShowTool);
|
||||
impl_from_backend!(TicketCommentTool);
|
||||
impl_from_backend!(TicketPlanTool);
|
||||
impl_from_backend!(TicketDecisionTool);
|
||||
impl_from_backend!(TicketImplementationReportTool);
|
||||
impl_from_backend!(TicketReviewTool);
|
||||
impl_from_backend!(TicketIntakeReadyTool);
|
||||
impl_from_backend!(TicketQueueTool);
|
||||
@@ -1768,6 +1797,12 @@ pub fn ticket_tools(backend: impl Into<TicketToolBackend>) -> Vec<ToolDefinition
|
||||
tool_definition::<TicketListTool>("TicketList", backend.clone()),
|
||||
tool_definition::<TicketShowTool>("TicketShow", backend.clone()),
|
||||
tool_definition::<TicketCommentTool>("TicketComment", backend.clone()),
|
||||
tool_definition::<TicketPlanTool>("TicketPlan", backend.clone()),
|
||||
tool_definition::<TicketDecisionTool>("TicketDecision", backend.clone()),
|
||||
tool_definition::<TicketImplementationReportTool>(
|
||||
"TicketImplementationReport",
|
||||
backend.clone(),
|
||||
),
|
||||
tool_definition::<TicketReviewTool>("TicketReview", backend.clone()),
|
||||
tool_definition::<TicketIntakeReadyTool>("TicketIntakeReady", backend.clone()),
|
||||
tool_definition::<TicketQueueTool>("TicketQueue", backend.clone()),
|
||||
@@ -1841,6 +1876,9 @@ mod tests {
|
||||
"TicketCreate",
|
||||
"TicketEditItem",
|
||||
"TicketComment",
|
||||
"TicketPlan",
|
||||
"TicketDecision",
|
||||
"TicketImplementationReport",
|
||||
"TicketReview",
|
||||
"TicketIntakeReady",
|
||||
"TicketQueue",
|
||||
@@ -2338,16 +2376,15 @@ mod tests {
|
||||
let temp = TempDir::new().unwrap();
|
||||
let backend = backend(&temp);
|
||||
let created = backend.create(NewTicket::new("Flow Tool")).unwrap();
|
||||
let comment = tool_by_name(backend.clone(), "TicketComment");
|
||||
let report = tool_by_name(backend.clone(), "TicketImplementationReport");
|
||||
let review = tool_by_name(backend.clone(), "TicketReview");
|
||||
let close = tool_by_name(backend.clone(), "TicketClose");
|
||||
let doctor = tool_by_name(backend.clone(), "TicketDoctor");
|
||||
|
||||
comment
|
||||
report
|
||||
.execute(
|
||||
&json!({
|
||||
"ticket": created.id.clone(),
|
||||
"role": "implementation_report",
|
||||
"body": "Implemented."
|
||||
})
|
||||
.to_string(),
|
||||
@@ -2807,6 +2844,38 @@ mod tests {
|
||||
assert!(edit_schema.contains("old_string"));
|
||||
assert!(edit_schema.contains("new_string"));
|
||||
assert!(edit_schema.contains("replace_all"));
|
||||
for name in [
|
||||
"TicketCreate",
|
||||
"TicketEditItem",
|
||||
"TicketComment",
|
||||
"TicketPlan",
|
||||
"TicketDecision",
|
||||
"TicketImplementationReport",
|
||||
"TicketReview",
|
||||
"TicketIntakeReady",
|
||||
"TicketQueue",
|
||||
"TicketRelationRecord",
|
||||
"TicketOrchestrationPlanRecord",
|
||||
] {
|
||||
let schema = tools
|
||||
.iter()
|
||||
.map(|definition| definition().0)
|
||||
.find(|meta| meta.name == name)
|
||||
.unwrap()
|
||||
.input_schema;
|
||||
let properties = schema["properties"].as_object().unwrap();
|
||||
assert!(!properties.contains_key("author"), "{name} exposes author");
|
||||
assert!(
|
||||
!properties.contains_key("queued_by"),
|
||||
"{name} exposes queued_by"
|
||||
);
|
||||
if matches!(
|
||||
name,
|
||||
"TicketComment" | "TicketPlan" | "TicketDecision" | "TicketImplementationReport"
|
||||
) {
|
||||
assert!(!properties.contains_key("role"), "{name} exposes role");
|
||||
}
|
||||
}
|
||||
let names = tools
|
||||
.into_iter()
|
||||
.map(|definition| definition().0)
|
||||
|
||||
@@ -172,10 +172,29 @@ pub struct WorkingDirectoryStatus {
|
||||
pub summary: WorkingDirectorySummary,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct WorkspaceApiRef {
|
||||
pub workspace_id: String,
|
||||
pub base_url: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub runtime_id: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub access_token: Option<String>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for WorkspaceApiRef {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
formatter
|
||||
.debug_struct("WorkspaceApiRef")
|
||||
.field("workspace_id", &self.workspace_id)
|
||||
.field("base_url", &self.base_url)
|
||||
.field("runtime_id", &self.runtime_id)
|
||||
.field(
|
||||
"access_token",
|
||||
&self.access_token.as_ref().map(|_| "[redacted]"),
|
||||
)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
/// Canonical Runtime Worker creation request.
|
||||
@@ -189,6 +208,10 @@ pub struct WorkspaceApiRef {
|
||||
/// summarized without exposing raw host paths.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct CreateWorkerRequest {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub idempotency_key: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub idempotency_fingerprint: Option<String>,
|
||||
pub profile: ProfileSelector,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub display_name: Option<String>,
|
||||
|
||||
@@ -1153,6 +1153,8 @@ mod tests {
|
||||
request.workspace_api = Some(WorkspaceApiRef {
|
||||
workspace_id: workspace_id.to_string(),
|
||||
base_url: format!("https://workspace.example/{workspace_id}"),
|
||||
runtime_id: None,
|
||||
access_token: None,
|
||||
});
|
||||
request
|
||||
}
|
||||
@@ -1410,6 +1412,8 @@ mod tests {
|
||||
let profile = ProfileSelector::Builtin("builtin:coder".to_string());
|
||||
let bundle = test_bundle(profile.clone());
|
||||
CreateWorkerRequest {
|
||||
idempotency_key: None,
|
||||
idempotency_fingerprint: None,
|
||||
profile,
|
||||
display_name: None,
|
||||
profile_source: crate::catalog::ProfileSourceArchiveSource::Http {
|
||||
@@ -1810,6 +1814,8 @@ mod ws_tests {
|
||||
fn ws_create_request() -> CreateWorkerRequest {
|
||||
let bundle = ws_test_bundle(ProfileSelector::Builtin("builtin:companion".to_string()));
|
||||
CreateWorkerRequest {
|
||||
idempotency_key: None,
|
||||
idempotency_fingerprint: None,
|
||||
profile: ProfileSelector::Builtin("builtin:companion".to_string()),
|
||||
display_name: None,
|
||||
profile_source: crate::catalog::ProfileSourceArchiveSource::Http {
|
||||
|
||||
@@ -354,6 +354,11 @@ impl Runtime {
|
||||
request: CreateWorkerRequest,
|
||||
scope: Option<&RuntimeWorkspaceScope>,
|
||||
) -> Result<WorkerDetail, RuntimeError> {
|
||||
if request.idempotency_key.is_some() != request.idempotency_fingerprint.is_some() {
|
||||
return Err(RuntimeError::InvalidRequest(
|
||||
"idempotency_key and idempotency_fingerprint must be provided together".to_string(),
|
||||
));
|
||||
}
|
||||
let (backend, worker_ref, spawn_request) = {
|
||||
let mut state = self.lock()?;
|
||||
state.ensure_running()?;
|
||||
@@ -365,6 +370,20 @@ impl Runtime {
|
||||
if let Some(scope) = scope {
|
||||
state.ensure_workspace_owner(scope, true)?;
|
||||
};
|
||||
if let Some(idempotency_key) = request.idempotency_key.as_deref() {
|
||||
let workspace_id = scope.map(|scope| scope.workspace_id.as_str());
|
||||
if let Some(existing) = state.workers.values().find(|record| {
|
||||
record.workspace_id.as_deref() == workspace_id
|
||||
&& record.request.idempotency_key.as_deref() == Some(idempotency_key)
|
||||
}) {
|
||||
if existing.request.idempotency_fingerprint != request.idempotency_fingerprint {
|
||||
return Err(RuntimeError::InvalidRequest(format!(
|
||||
"worker creation idempotency key {idempotency_key} was already used with different input"
|
||||
)));
|
||||
}
|
||||
return Ok(existing.detail());
|
||||
}
|
||||
}
|
||||
state.validate_worker_config_boundary(&request)?;
|
||||
if let Some(working_directory_id) = requested_primary_workdir_id(&request) {
|
||||
if let Some(owner_worker_id) =
|
||||
@@ -2108,7 +2127,9 @@ fn input_protocol_event(input: &WorkerInput) -> protocol::Event {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::catalog::{ConfigBundleRef, ProfileSelector, WorkspaceApiRef};
|
||||
use crate::catalog::{
|
||||
ConfigBundleRef, ProfileSelector, WorkingDirectoryClaim, WorkspaceApiRef,
|
||||
};
|
||||
use crate::config_bundle::{
|
||||
ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration,
|
||||
ConfigDeclarationKind, ConfigProfileDescriptor,
|
||||
@@ -2126,6 +2147,8 @@ mod tests {
|
||||
let profile = ProfileSelector::Builtin("builtin:coder".to_string());
|
||||
let bundle = test_bundle_for_profile(profile.clone());
|
||||
CreateWorkerRequest {
|
||||
idempotency_key: None,
|
||||
idempotency_fingerprint: None,
|
||||
profile,
|
||||
display_name: None,
|
||||
profile_source: crate::catalog::ProfileSourceArchiveSource::Http {
|
||||
@@ -2161,6 +2184,8 @@ mod tests {
|
||||
request.workspace_api = Some(WorkspaceApiRef {
|
||||
workspace_id: workspace_id.to_string(),
|
||||
base_url: format!("https://workspace.example/{workspace_id}"),
|
||||
runtime_id: None,
|
||||
access_token: None,
|
||||
});
|
||||
request
|
||||
}
|
||||
@@ -2612,6 +2637,33 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_worker_idempotency_reuses_worker_and_rejects_different_input() {
|
||||
let runtime = runtime_with_backend();
|
||||
let mut request = task_request("idempotent");
|
||||
request.idempotency_key = Some("operation-1".to_string());
|
||||
request.idempotency_fingerprint = Some("sha256:input-1".to_string());
|
||||
request.working_directory = Some(WorkingDirectoryClaim {
|
||||
working_directory_id: "workdir-idempotent".to_string(),
|
||||
relative_cwd: None,
|
||||
});
|
||||
|
||||
let first = runtime.create_worker(request.clone()).unwrap();
|
||||
let workdir_count_after_first = runtime.list_working_directories().unwrap().len();
|
||||
let replayed = runtime.create_worker(request.clone()).unwrap();
|
||||
assert_eq!(replayed.worker_ref, first.worker_ref);
|
||||
assert_eq!(runtime.list_workers().unwrap().len(), 1);
|
||||
assert_eq!(
|
||||
runtime.list_working_directories().unwrap().len(),
|
||||
workdir_count_after_first
|
||||
);
|
||||
|
||||
request.idempotency_fingerprint = Some("sha256:different".to_string());
|
||||
let error = runtime.create_worker(request).unwrap_err();
|
||||
assert!(matches!(error, RuntimeError::InvalidRequest(_)));
|
||||
assert_eq!(runtime.list_workers().unwrap().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_worker_rejects_system_initial_input_without_persisting_worker() {
|
||||
let runtime = runtime_with_backend();
|
||||
|
||||
@@ -23,6 +23,7 @@ use crate::execution::{
|
||||
WorkerExecutionRestoreRequest, WorkerExecutionResult, WorkerExecutionRunState,
|
||||
WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult,
|
||||
};
|
||||
use crate::identity::WorkerRef;
|
||||
use crate::interaction::{WorkerInput, WorkerInputKind};
|
||||
use crate::resource::{BackendResourceClient, ProfileSourceArchiveCache};
|
||||
use crate::working_directory::{
|
||||
@@ -40,8 +41,8 @@ use tokio::sync::broadcast;
|
||||
#[cfg(feature = "ws-server")]
|
||||
use worker::ipc::protocol_session::{live_log_entry_event, subscribe_worker_protocol_session};
|
||||
use worker::{
|
||||
PromptLoader, Worker, WorkerController, WorkerError, WorkerFilesystemAuthority, WorkerHandle,
|
||||
WorkerWorkspaceContext, WorkspaceClient, WorkspaceId,
|
||||
PromptLoader, RuntimeWorkspaceHttpClient, Worker, WorkerController, WorkerError,
|
||||
WorkerFilesystemAuthority, WorkerHandle, WorkerWorkspaceContext, WorkspaceId,
|
||||
};
|
||||
|
||||
const DEFAULT_BACKEND_ID: &str = "worker-crate";
|
||||
@@ -259,6 +260,7 @@ enum RuntimeWorkspaceBackendRef {
|
||||
Http {
|
||||
workspace_id: String,
|
||||
base_url: String,
|
||||
access_token: Option<String>,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -268,20 +270,29 @@ impl RuntimeWorkspaceBackendRef {
|
||||
return Self::Http {
|
||||
workspace_id: api.workspace_id.clone(),
|
||||
base_url: api.base_url.clone(),
|
||||
access_token: api.access_token.clone(),
|
||||
};
|
||||
}
|
||||
Self::None
|
||||
}
|
||||
|
||||
fn worker_context(&self) -> WorkerWorkspaceContext {
|
||||
fn worker_context(&self, worker_ref: &WorkerRef) -> WorkerWorkspaceContext {
|
||||
match self {
|
||||
Self::None => WorkerWorkspaceContext::no_workspace(),
|
||||
Self::Http {
|
||||
workspace_id,
|
||||
base_url,
|
||||
access_token,
|
||||
} => WorkerWorkspaceContext::with_client(
|
||||
WorkspaceId::new(workspace_id.clone()).ok(),
|
||||
WorkspaceClient::http(workspace_id.clone(), base_url.clone()),
|
||||
Arc::new(
|
||||
RuntimeWorkspaceHttpClient::new(
|
||||
workspace_id.clone(),
|
||||
base_url.clone(),
|
||||
worker_ref.worker_id.to_string(),
|
||||
)
|
||||
.with_access_token(access_token.clone()),
|
||||
),
|
||||
),
|
||||
}
|
||||
}
|
||||
@@ -368,7 +379,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
||||
.unwrap_or(WorkerFilesystemAuthority::None);
|
||||
let workspace_backend_ref =
|
||||
RuntimeWorkspaceBackendRef::from_worker_request(&request.request);
|
||||
let workspace_context = workspace_backend_ref.worker_context();
|
||||
let workspace_context = workspace_backend_ref.worker_context(&request.worker_ref);
|
||||
let selector = profile.as_ref();
|
||||
let archive = self
|
||||
.resolve_profile_source_archive(&request.request.profile_source)
|
||||
@@ -442,7 +453,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
||||
.unwrap_or(WorkerFilesystemAuthority::None);
|
||||
let workspace_backend_ref =
|
||||
RuntimeWorkspaceBackendRef::from_worker_request(&request.request);
|
||||
let workspace_context = workspace_backend_ref.worker_context();
|
||||
let workspace_context = workspace_backend_ref.worker_context(&request.worker_ref);
|
||||
let (manifest, loader) = Self::restore_fallback_manifest(&worker_name)?;
|
||||
|
||||
let store_dir = self.store_dir()?;
|
||||
@@ -1276,7 +1287,7 @@ mod tests {
|
||||
store_dir: PathBuf,
|
||||
worker_metadata_dir: PathBuf,
|
||||
observed_cwds: Arc<Mutex<Vec<PathBuf>>>,
|
||||
observed_workspace_clients: Arc<Mutex<Vec<WorkspaceClient>>>,
|
||||
observed_workspace_clients: Arc<Mutex<Vec<(String, Option<String>, bool)>>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -1325,11 +1336,13 @@ mod tests {
|
||||
.unwrap_or_else(|| self.cwd.clone());
|
||||
let workspace_backend_ref =
|
||||
RuntimeWorkspaceBackendRef::from_worker_request(&request.request);
|
||||
let workspace_context = workspace_backend_ref.worker_context();
|
||||
self.observed_workspace_clients
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(workspace_context.client().clone());
|
||||
let workspace_context = workspace_backend_ref.worker_context(&request.worker_ref);
|
||||
let workspace_client = workspace_context.client();
|
||||
self.observed_workspace_clients.lock().unwrap().push((
|
||||
workspace_client.kind().to_string(),
|
||||
workspace_client.workspace_id().map(str::to_string),
|
||||
workspace_client.is_available(),
|
||||
));
|
||||
let scope = Scope::writable(&scope_root).map_err(|err| err.to_string())?;
|
||||
let worker = Worker::new(
|
||||
manifest,
|
||||
@@ -1438,6 +1451,8 @@ mod tests {
|
||||
fn create_request(_name: &str) -> CreateWorkerRequest {
|
||||
let bundle = test_bundle();
|
||||
CreateWorkerRequest {
|
||||
idempotency_key: None,
|
||||
idempotency_fingerprint: None,
|
||||
profile: ProfileSelector::Builtin("builtin:companion".to_string()),
|
||||
display_name: None,
|
||||
profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded {
|
||||
@@ -1673,6 +1688,8 @@ mod tests {
|
||||
request.workspace_api = Some(crate::catalog::WorkspaceApiRef {
|
||||
workspace_id: "ws-test".to_string(),
|
||||
base_url: "http://127.0.0.1:3999".to_string(),
|
||||
runtime_id: None,
|
||||
access_token: None,
|
||||
});
|
||||
let detail = runtime.create_worker(request).unwrap();
|
||||
|
||||
@@ -1704,7 +1721,11 @@ mod tests {
|
||||
assert!(observed_cwds.lock().unwrap().is_empty());
|
||||
assert_eq!(
|
||||
observed_workspace_clients.lock().unwrap().as_slice(),
|
||||
&[WorkspaceClient::http("ws-test", "http://127.0.0.1:3999")]
|
||||
&[(
|
||||
"runtime-http-proxy".to_string(),
|
||||
Some("ws-test".to_string()),
|
||||
true,
|
||||
)]
|
||||
);
|
||||
let names = captured_tool_names(&client, 0);
|
||||
for forbidden in core_filesystem_tool_names() {
|
||||
@@ -1781,9 +1802,7 @@ mod tests {
|
||||
assert!(cwd.join("README.md").exists());
|
||||
assert_eq!(
|
||||
observed_workspace_clients.lock().unwrap().as_slice(),
|
||||
&[WorkspaceClient::Unavailable {
|
||||
reason: "no workspace configured".to_string()
|
||||
}]
|
||||
&[("unavailable".to_string(), None, false)]
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -26,7 +26,7 @@ use crate::shutdown_after_idle::{
|
||||
use crate::spawn::comm_tools::{read_worker_output_tool, send_to_worker_tool, stop_worker_tool};
|
||||
use crate::spawn::registry::SpawnedWorkerRegistry;
|
||||
use crate::spawn::tool::spawn_worker_tool;
|
||||
use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult, WorkspaceClient};
|
||||
use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult};
|
||||
use protocol::{
|
||||
AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment,
|
||||
TurnResult, WorkerStatus,
|
||||
@@ -627,21 +627,16 @@ where
|
||||
// Ticket tools are typed operations over the current workspace Ticket backend.
|
||||
// Workspace access must be authority-bound to the Backend Workspace API; the
|
||||
// Worker must not fall back to a local `.yoi/tickets` store.
|
||||
let ticket_backend = match worker.workspace_client() {
|
||||
WorkspaceClient::Http {
|
||||
workspace_id,
|
||||
base_url,
|
||||
} => crate::feature::builtin::ticket::TicketFeatureBackend::WorkspaceHttp {
|
||||
workspace_id: workspace_id.clone(),
|
||||
base_url: base_url.clone(),
|
||||
},
|
||||
_ => {
|
||||
let workspace_client = worker.workspace_client_handle();
|
||||
if !workspace_client.is_available() || workspace_client.workspace_id().is_none() {
|
||||
return Err(std::io::Error::new(
|
||||
std::io::ErrorKind::InvalidInput,
|
||||
"ticket tools require Backend Workspace API authority",
|
||||
));
|
||||
}
|
||||
};
|
||||
let ticket_backend = crate::feature::builtin::ticket::TicketFeatureBackend::WorkspaceClient(
|
||||
workspace_client,
|
||||
);
|
||||
feature_registry.add_module(
|
||||
crate::feature::builtin::ticket::ticket_tools_feature_with_backend(
|
||||
ticket_backend,
|
||||
@@ -668,21 +663,16 @@ where
|
||||
}
|
||||
|
||||
{
|
||||
let workspace_client = worker.workspace_client().clone();
|
||||
let workspace_client = worker.workspace_client_handle();
|
||||
let engine = worker.engine_mut();
|
||||
|
||||
// Objective tools expose read-only project Objective context through the
|
||||
// Backend Workspace API. Workers must not guess local `.yoi/objectives`
|
||||
// paths or read Objective files directly.
|
||||
if feature_config.objective.enabled {
|
||||
if let WorkspaceClient::Http {
|
||||
workspace_id,
|
||||
base_url,
|
||||
} = &workspace_client
|
||||
{
|
||||
if workspace_client.is_available() && workspace_client.workspace_id().is_some() {
|
||||
for definition in crate::feature::builtin::objective::workspace_http_objective_tools(
|
||||
workspace_id.clone(),
|
||||
base_url.clone(),
|
||||
workspace_client.clone(),
|
||||
) {
|
||||
engine.register_tool(definition);
|
||||
}
|
||||
@@ -705,20 +695,14 @@ where
|
||||
"[feature.memory].enabled = true requires a [memory] configuration section",
|
||||
)
|
||||
})?;
|
||||
if let WorkspaceClient::Http {
|
||||
workspace_id,
|
||||
base_url,
|
||||
} = workspace_client
|
||||
{
|
||||
if workspace_client.is_available() && workspace_client.workspace_id().is_some() {
|
||||
let definitions = if feature_config.memory.staging {
|
||||
crate::feature::builtin::memory::workspace_http_memory_consolidation_tools(
|
||||
workspace_id,
|
||||
base_url,
|
||||
workspace_client.clone(),
|
||||
)
|
||||
} else {
|
||||
crate::feature::builtin::memory::workspace_http_memory_tools(
|
||||
workspace_id,
|
||||
base_url,
|
||||
workspace_client.clone(),
|
||||
)
|
||||
};
|
||||
for definition in definitions {
|
||||
|
||||
@@ -20,27 +20,25 @@ use schemars::JsonSchema;
|
||||
use serde::de::DeserializeOwned;
|
||||
use serde_json::json;
|
||||
|
||||
use crate::worker::WorkspaceClient;
|
||||
use crate::worker::{
|
||||
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
|
||||
};
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct WorkspaceHttpMemoryBackend {
|
||||
workspace_id: String,
|
||||
base_url: String,
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
}
|
||||
|
||||
impl WorkspaceHttpMemoryBackend {
|
||||
pub fn new(workspace_id: impl Into<String>, base_url: impl Into<String>) -> Self {
|
||||
Self {
|
||||
workspace_id: workspace_id.into(),
|
||||
base_url: base_url.into(),
|
||||
}
|
||||
pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
|
||||
Self { client }
|
||||
}
|
||||
|
||||
pub async fn execute_operation(
|
||||
&self,
|
||||
operation: MemoryBackendOperation,
|
||||
) -> Result<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
|
||||
execute_http_memory_backend(&self.workspace_id, &self.base_url, operation).await
|
||||
execute_memory_backend(self.client.as_ref(), operation).await
|
||||
}
|
||||
|
||||
async fn execute(&self, operation: MemoryBackendOperation) -> Result<ToolOutput, ToolError> {
|
||||
@@ -59,7 +57,7 @@ pub enum WorkspaceMemoryBackendError {
|
||||
#[error("workspace memory backend is unavailable: {reason}")]
|
||||
Unavailable { reason: String },
|
||||
#[error("workspace memory backend request failed: {0}")]
|
||||
Request(#[from] reqwest::Error),
|
||||
Request(#[from] WorkspaceClientError),
|
||||
#[error("workspace memory backend returned HTTP {status}: {body}")]
|
||||
Http {
|
||||
status: reqwest::StatusCode,
|
||||
@@ -71,73 +69,49 @@ pub enum WorkspaceMemoryBackendError {
|
||||
Backend(String),
|
||||
}
|
||||
|
||||
impl WorkspaceClient {
|
||||
impl dyn WorkspaceClient + '_ {
|
||||
pub async fn execute_memory_backend_operation(
|
||||
&self,
|
||||
operation: MemoryBackendOperation,
|
||||
) -> Result<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
|
||||
match self {
|
||||
WorkspaceClient::Http {
|
||||
workspace_id,
|
||||
base_url,
|
||||
} => execute_http_memory_backend(workspace_id, base_url, operation).await,
|
||||
WorkspaceClient::Available { kind } => Err(WorkspaceMemoryBackendError::Unavailable {
|
||||
reason: format!(
|
||||
"workspace client kind `{kind}` does not expose the Backend Workspace API"
|
||||
),
|
||||
}),
|
||||
WorkspaceClient::Unavailable { reason } => {
|
||||
Err(WorkspaceMemoryBackendError::Unavailable {
|
||||
reason: reason.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
execute_memory_backend(self, operation).await
|
||||
}
|
||||
|
||||
pub async fn request_memory_staging_consolidation(
|
||||
&self,
|
||||
operation: MemoryConsolidateStagingOperation,
|
||||
) -> Result<MemoryConsolidationOutput, WorkspaceMemoryBackendError> {
|
||||
match self {
|
||||
WorkspaceClient::Http {
|
||||
workspace_id,
|
||||
base_url,
|
||||
} => execute_http_memory_consolidation(workspace_id, base_url, operation).await,
|
||||
WorkspaceClient::Available { kind } => Err(WorkspaceMemoryBackendError::Unavailable {
|
||||
reason: format!(
|
||||
"workspace client kind `{kind}` does not expose the Backend Workspace API"
|
||||
),
|
||||
}),
|
||||
WorkspaceClient::Unavailable { reason } => {
|
||||
Err(WorkspaceMemoryBackendError::Unavailable {
|
||||
reason: reason.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
execute_memory_consolidation(self, operation).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_http_memory_backend(
|
||||
workspace_id: &str,
|
||||
base_url: &str,
|
||||
async fn execute_memory_backend(
|
||||
client: &dyn WorkspaceClient,
|
||||
operation: MemoryBackendOperation,
|
||||
) -> Result<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
|
||||
let url = format!(
|
||||
"{}/api/w/{}/memory/backend",
|
||||
base_url.trim_end_matches('/'),
|
||||
workspace_id
|
||||
);
|
||||
let response = reqwest::Client::new()
|
||||
.post(url)
|
||||
.json(&operation)
|
||||
.send()
|
||||
.await?;
|
||||
let status = response.status();
|
||||
let body = response.text().await?;
|
||||
if !status.is_success() {
|
||||
return Err(WorkspaceMemoryBackendError::Http { status, body });
|
||||
let workspace_id =
|
||||
client
|
||||
.workspace_id()
|
||||
.ok_or_else(|| WorkspaceMemoryBackendError::Unavailable {
|
||||
reason: format!(
|
||||
"workspace client kind `{}` has no workspace id",
|
||||
client.kind()
|
||||
),
|
||||
})?;
|
||||
let response = client.execute(WorkspaceRequest::json(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!("/api/w/{workspace_id}/memory/backend"),
|
||||
serde_json::to_string(&operation)?,
|
||||
))?;
|
||||
let status = reqwest::StatusCode::from_u16(response.status)
|
||||
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
|
||||
if !response.is_success() {
|
||||
return Err(WorkspaceMemoryBackendError::Http {
|
||||
status,
|
||||
body: response.body,
|
||||
});
|
||||
}
|
||||
match serde_json::from_str::<MemoryBackendHttpResponse>(&body)? {
|
||||
match serde_json::from_str::<MemoryBackendHttpResponse>(&response.body)? {
|
||||
MemoryBackendHttpResponse::Ok { result } => Ok(result),
|
||||
MemoryBackendHttpResponse::Error { message } => {
|
||||
Err(WorkspaceMemoryBackendError::Backend(message))
|
||||
@@ -145,34 +119,37 @@ async fn execute_http_memory_backend(
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_http_memory_consolidation(
|
||||
workspace_id: &str,
|
||||
base_url: &str,
|
||||
async fn execute_memory_consolidation(
|
||||
client: &dyn WorkspaceClient,
|
||||
operation: MemoryConsolidateStagingOperation,
|
||||
) -> Result<MemoryConsolidationOutput, WorkspaceMemoryBackendError> {
|
||||
let url = format!(
|
||||
"{}/api/w/{}/memory/consolidation",
|
||||
base_url.trim_end_matches('/'),
|
||||
workspace_id
|
||||
);
|
||||
let response = reqwest::Client::new()
|
||||
.post(url)
|
||||
.json(&operation)
|
||||
.send()
|
||||
.await?;
|
||||
let status = response.status();
|
||||
let body = response.text().await?;
|
||||
if !status.is_success() {
|
||||
return Err(WorkspaceMemoryBackendError::Http { status, body });
|
||||
let workspace_id =
|
||||
client
|
||||
.workspace_id()
|
||||
.ok_or_else(|| WorkspaceMemoryBackendError::Unavailable {
|
||||
reason: format!(
|
||||
"workspace client kind `{}` has no workspace id",
|
||||
client.kind()
|
||||
),
|
||||
})?;
|
||||
let response = client.execute(WorkspaceRequest::json(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!("/api/w/{workspace_id}/memory/consolidation"),
|
||||
serde_json::to_string(&operation)?,
|
||||
))?;
|
||||
let status = reqwest::StatusCode::from_u16(response.status)
|
||||
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
|
||||
if !response.is_success() {
|
||||
return Err(WorkspaceMemoryBackendError::Http {
|
||||
status,
|
||||
body: response.body,
|
||||
});
|
||||
}
|
||||
serde_json::from_str::<MemoryConsolidationOutput>(&body).map_err(Into::into)
|
||||
serde_json::from_str::<MemoryConsolidationOutput>(&response.body).map_err(Into::into)
|
||||
}
|
||||
|
||||
pub fn workspace_http_memory_tools(
|
||||
workspace_id: impl Into<String>,
|
||||
base_url: impl Into<String>,
|
||||
) -> Vec<ToolDefinition> {
|
||||
let backend = WorkspaceHttpMemoryBackend::new(workspace_id, base_url);
|
||||
pub fn workspace_http_memory_tools(client: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
|
||||
let backend = WorkspaceHttpMemoryBackend::new(client);
|
||||
vec![
|
||||
memory_tool(
|
||||
"MemoryReadDocument",
|
||||
@@ -215,13 +192,10 @@ pub fn workspace_http_memory_tools(
|
||||
}
|
||||
|
||||
pub fn workspace_http_memory_consolidation_tools(
|
||||
workspace_id: impl Into<String>,
|
||||
base_url: impl Into<String>,
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
) -> Vec<ToolDefinition> {
|
||||
let workspace_id = workspace_id.into();
|
||||
let base_url = base_url.into();
|
||||
let mut tools = workspace_http_memory_tools(workspace_id.clone(), base_url.clone());
|
||||
let backend = WorkspaceHttpMemoryBackend::new(workspace_id, base_url);
|
||||
let mut tools = workspace_http_memory_tools(client.clone());
|
||||
let backend = WorkspaceHttpMemoryBackend::new(client);
|
||||
tools.extend([
|
||||
memory_tool(
|
||||
"MemoryStagingList",
|
||||
@@ -370,6 +344,14 @@ mod tests {
|
||||
use super::*;
|
||||
use llm_engine::tool::ToolDefinition;
|
||||
|
||||
fn test_client() -> Arc<dyn WorkspaceClient> {
|
||||
Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new(
|
||||
"workspace",
|
||||
"http://backend",
|
||||
"test-worker",
|
||||
))
|
||||
}
|
||||
|
||||
fn tool_names(definitions: Vec<ToolDefinition>) -> Vec<String> {
|
||||
let mut names = definitions
|
||||
.into_iter()
|
||||
@@ -390,10 +372,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn normal_workspace_memory_tools_do_not_include_staging_tools() {
|
||||
let names = tool_names(workspace_http_memory_tools(
|
||||
"workspace".to_string(),
|
||||
"http://backend".to_string(),
|
||||
));
|
||||
let names = tool_names(workspace_http_memory_tools(test_client()));
|
||||
|
||||
assert!(names.contains(&"MemoryQuery".to_string()));
|
||||
assert!(names.contains(&"MemoryReadDocument".to_string()));
|
||||
@@ -410,7 +389,7 @@ mod tests {
|
||||
#[test]
|
||||
fn document_update_schema_is_edit_like_and_staging_close_has_no_legacy_kinds() {
|
||||
let update_schema = tool_meta(
|
||||
workspace_http_memory_tools("workspace".to_string(), "http://backend".to_string()),
|
||||
workspace_http_memory_tools(test_client()),
|
||||
"MemoryUpdateDocument",
|
||||
);
|
||||
assert_eq!(
|
||||
@@ -423,10 +402,7 @@ mod tests {
|
||||
assert!(update_schema["properties"].get("body_md").is_none());
|
||||
|
||||
let close_schema_text = tool_meta(
|
||||
workspace_http_memory_consolidation_tools(
|
||||
"workspace".to_string(),
|
||||
"http://backend".to_string(),
|
||||
),
|
||||
workspace_http_memory_consolidation_tools(test_client()),
|
||||
"MemoryStagingClose",
|
||||
)
|
||||
.to_string();
|
||||
@@ -440,10 +416,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn consolidation_workspace_memory_tools_include_staging_tools() {
|
||||
let names = tool_names(workspace_http_memory_consolidation_tools(
|
||||
"workspace".to_string(),
|
||||
"http://backend".to_string(),
|
||||
));
|
||||
let names = tool_names(workspace_http_memory_consolidation_tools(test_client()));
|
||||
|
||||
assert!(names.contains(&"MemoryQuery".to_string()));
|
||||
assert!(names.contains(&"MemoryReadDocument".to_string()));
|
||||
|
||||
@@ -14,26 +14,27 @@ use llm_engine::tool::{
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::json;
|
||||
|
||||
use crate::worker::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod};
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct WorkspaceHttpObjectiveBackend {
|
||||
workspace_id: String,
|
||||
base_url: String,
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
}
|
||||
|
||||
impl WorkspaceHttpObjectiveBackend {
|
||||
pub fn new(workspace_id: impl Into<String>, base_url: impl Into<String>) -> Self {
|
||||
Self {
|
||||
workspace_id: workspace_id.into(),
|
||||
base_url: base_url.into().trim_end_matches('/').to_string(),
|
||||
}
|
||||
pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
|
||||
Self { client }
|
||||
}
|
||||
|
||||
async fn list(&self, input: ObjectiveListInput) -> Result<ToolOutput, ToolError> {
|
||||
let mut url = format!("{}/api/w/{}/objectives", self.base_url, self.workspace_id);
|
||||
let mut url = format!(
|
||||
"/api/w/{}/objectives",
|
||||
self.client.workspace_id().unwrap_or_default()
|
||||
);
|
||||
if let Some(limit) = input.limit {
|
||||
url.push_str(&format!("?limit={}", limit.min(1000)));
|
||||
}
|
||||
let response = get_json::<ObjectiveListResponse>(&url)
|
||||
let response = get_json::<ObjectiveListResponse>(self.client.as_ref(), &url)
|
||||
.await
|
||||
.map_err(backend_error)?;
|
||||
let count = response.items.len();
|
||||
@@ -46,7 +47,7 @@ impl WorkspaceHttpObjectiveBackend {
|
||||
async fn show(&self, input: ObjectiveShowInput) -> Result<ToolOutput, ToolError> {
|
||||
let id = validate_id(&input.id, "ObjectiveShow")?;
|
||||
let url = self.objective_url(id);
|
||||
let response = get_json::<ObjectiveDetail>(&url)
|
||||
let response = get_json::<ObjectiveDetail>(self.client.as_ref(), &url)
|
||||
.await
|
||||
.map_err(backend_error)?;
|
||||
Ok(objective_output(
|
||||
@@ -61,9 +62,16 @@ impl WorkspaceHttpObjectiveBackend {
|
||||
"ObjectiveCreate requires non-empty title".to_string(),
|
||||
));
|
||||
}
|
||||
let url = format!("{}/api/w/{}/objectives", self.base_url, self.workspace_id);
|
||||
let response =
|
||||
send_json::<ObjectiveCreateInput, ObjectiveDetail>(reqwest::Method::POST, &url, &input)
|
||||
let url = format!(
|
||||
"/api/w/{}/objectives",
|
||||
self.client.workspace_id().unwrap_or_default()
|
||||
);
|
||||
let response = send_json::<ObjectiveCreateInput, ObjectiveDetail>(
|
||||
self.client.as_ref(),
|
||||
reqwest::Method::POST,
|
||||
&url,
|
||||
&input,
|
||||
)
|
||||
.await
|
||||
.map_err(backend_error)?;
|
||||
Ok(objective_output(
|
||||
@@ -86,8 +94,12 @@ impl WorkspaceHttpObjectiveBackend {
|
||||
new_string: input.new_string,
|
||||
replace_all: input.replace_all,
|
||||
};
|
||||
let response =
|
||||
send_json::<ObjectiveEditRequest, ObjectiveDetail>(reqwest::Method::PATCH, &url, &body)
|
||||
let response = send_json::<ObjectiveEditRequest, ObjectiveDetail>(
|
||||
self.client.as_ref(),
|
||||
reqwest::Method::PATCH,
|
||||
&url,
|
||||
&body,
|
||||
)
|
||||
.await
|
||||
.map_err(backend_error)?;
|
||||
Ok(objective_output(
|
||||
@@ -105,6 +117,7 @@ impl WorkspaceHttpObjectiveBackend {
|
||||
}
|
||||
let url = format!("{}/state", self.objective_url(id));
|
||||
let response = send_json::<ObjectiveSetStateRequest, ObjectiveDetail>(
|
||||
self.client.as_ref(),
|
||||
reqwest::Method::POST,
|
||||
&url,
|
||||
&ObjectiveSetStateRequest { state: input.state },
|
||||
@@ -122,6 +135,7 @@ impl WorkspaceHttpObjectiveBackend {
|
||||
let ticket_id = validate_id(&input.ticket_id, "ObjectiveLinkTicket")?;
|
||||
let url = format!("{}/ticket-links", self.objective_url(id));
|
||||
let response = send_json::<ObjectiveLinkTicketRequest, ObjectiveDetail>(
|
||||
self.client.as_ref(),
|
||||
reqwest::Method::POST,
|
||||
&url,
|
||||
&ObjectiveLinkTicketRequest {
|
||||
@@ -143,7 +157,7 @@ impl WorkspaceHttpObjectiveBackend {
|
||||
let id = validate_id(&input.id, "ObjectiveUnlinkTicket")?;
|
||||
let ticket_id = validate_id(&input.ticket_id, "ObjectiveUnlinkTicket")?;
|
||||
let url = format!("{}/ticket-links/{}", self.objective_url(id), ticket_id);
|
||||
let response = delete_json::<ObjectiveDetail>(&url)
|
||||
let response = delete_json::<ObjectiveDetail>(self.client.as_ref(), &url)
|
||||
.await
|
||||
.map_err(backend_error)?;
|
||||
Ok(objective_output(
|
||||
@@ -153,17 +167,15 @@ impl WorkspaceHttpObjectiveBackend {
|
||||
}
|
||||
|
||||
fn objective_url(&self, id: &str) -> String {
|
||||
format!(
|
||||
"{}/api/w/{}/objectives/{}",
|
||||
self.base_url, self.workspace_id, id
|
||||
)
|
||||
let workspace_id = self.client.workspace_id().unwrap_or_default();
|
||||
format!("/api/w/{workspace_id}/objectives/{id}")
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum WorkspaceObjectiveBackendError {
|
||||
#[error("workspace objective backend request failed: {0}")]
|
||||
Request(#[from] reqwest::Error),
|
||||
Request(#[from] crate::worker::WorkspaceClientError),
|
||||
#[error("workspace objective backend returned HTTP {status}: {body}")]
|
||||
Http {
|
||||
status: reqwest::StatusCode,
|
||||
@@ -182,41 +194,55 @@ fn backend_error(error: WorkspaceObjectiveBackendError) -> ToolError {
|
||||
}
|
||||
|
||||
async fn get_json<T: for<'de> Deserialize<'de>>(
|
||||
url: &str,
|
||||
client: &dyn WorkspaceClient,
|
||||
path: &str,
|
||||
) -> Result<T, WorkspaceObjectiveBackendError> {
|
||||
let response = reqwest::Client::new().get(url).send().await?;
|
||||
decode_response(response).await
|
||||
decode_response(client.execute(WorkspaceRequest::get(path))?)
|
||||
}
|
||||
|
||||
async fn send_json<B: Serialize, T: for<'de> Deserialize<'de>>(
|
||||
client: &dyn WorkspaceClient,
|
||||
method: reqwest::Method,
|
||||
url: &str,
|
||||
path: &str,
|
||||
body: &B,
|
||||
) -> Result<T, WorkspaceObjectiveBackendError> {
|
||||
let response = reqwest::Client::new()
|
||||
.request(method, url)
|
||||
.json(body)
|
||||
.send()
|
||||
.await?;
|
||||
decode_response(response).await
|
||||
let method = match method {
|
||||
reqwest::Method::POST => WorkspaceRequestMethod::Post,
|
||||
reqwest::Method::PUT => WorkspaceRequestMethod::Put,
|
||||
reqwest::Method::PATCH => WorkspaceRequestMethod::Patch,
|
||||
reqwest::Method::DELETE => WorkspaceRequestMethod::Delete,
|
||||
_ => WorkspaceRequestMethod::Get,
|
||||
};
|
||||
decode_response(client.execute(WorkspaceRequest::json(
|
||||
method,
|
||||
path,
|
||||
serde_json::to_string(body)?,
|
||||
))?)
|
||||
}
|
||||
|
||||
async fn delete_json<T: for<'de> Deserialize<'de>>(
|
||||
url: &str,
|
||||
client: &dyn WorkspaceClient,
|
||||
path: &str,
|
||||
) -> Result<T, WorkspaceObjectiveBackendError> {
|
||||
let response = reqwest::Client::new().delete(url).send().await?;
|
||||
decode_response(response).await
|
||||
decode_response(client.execute(WorkspaceRequest {
|
||||
method: WorkspaceRequestMethod::Delete,
|
||||
path: path.to_string(),
|
||||
body: None,
|
||||
})?)
|
||||
}
|
||||
|
||||
async fn decode_response<T: for<'de> Deserialize<'de>>(
|
||||
response: reqwest::Response,
|
||||
fn decode_response<T: for<'de> Deserialize<'de>>(
|
||||
response: crate::worker::WorkspaceResponse,
|
||||
) -> Result<T, WorkspaceObjectiveBackendError> {
|
||||
let status = response.status();
|
||||
let body = response.text().await?;
|
||||
if !status.is_success() {
|
||||
return Err(WorkspaceObjectiveBackendError::Http { status, body });
|
||||
let status = reqwest::StatusCode::from_u16(response.status)
|
||||
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
|
||||
if !response.is_success() {
|
||||
return Err(WorkspaceObjectiveBackendError::Http {
|
||||
status,
|
||||
body: response.body,
|
||||
});
|
||||
}
|
||||
serde_json::from_str(&body).map_err(Into::into)
|
||||
serde_json::from_str(&response.body).map_err(Into::into)
|
||||
}
|
||||
|
||||
fn objective_output(summary: String, response: ObjectiveDetail) -> Result<ToolOutput, ToolError> {
|
||||
@@ -236,11 +262,8 @@ fn validate_id<'a>(id: &'a str, tool_name: &str) -> Result<&'a str, ToolError> {
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
pub fn workspace_http_objective_tools(
|
||||
workspace_id: impl Into<String>,
|
||||
base_url: impl Into<String>,
|
||||
) -> Vec<ToolDefinition> {
|
||||
let backend = WorkspaceHttpObjectiveBackend::new(workspace_id, base_url);
|
||||
pub fn workspace_http_objective_tools(client: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
|
||||
let backend = WorkspaceHttpObjectiveBackend::new(client);
|
||||
vec![
|
||||
objective_tool(
|
||||
"ObjectiveList",
|
||||
@@ -600,10 +623,13 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn workspace_http_objective_tools_include_objective_crud_tools() {
|
||||
let names = tool_names(workspace_http_objective_tools(
|
||||
"workspace".to_string(),
|
||||
"http://backend".to_string(),
|
||||
));
|
||||
let names = tool_names(workspace_http_objective_tools(Arc::new(
|
||||
crate::worker::RuntimeWorkspaceHttpClient::new(
|
||||
"workspace",
|
||||
"http://backend",
|
||||
"test-worker",
|
||||
),
|
||||
)));
|
||||
|
||||
assert_eq!(
|
||||
names,
|
||||
|
||||
@@ -29,7 +29,7 @@ const FINISH_EXTRACTION_DESCRIPTION: &str = "Finish the extract worker run after
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct SessionExploreState {
|
||||
view: Arc<SessionReferenceView>,
|
||||
workspace_client: WorkspaceClient,
|
||||
workspace_client: Arc<dyn WorkspaceClient>,
|
||||
source: SourceRef,
|
||||
extract_run_id: String,
|
||||
staged: Arc<Mutex<Vec<String>>>,
|
||||
@@ -39,7 +39,7 @@ pub(crate) struct SessionExploreState {
|
||||
impl SessionExploreState {
|
||||
pub(crate) fn new(
|
||||
view: SessionReferenceView,
|
||||
workspace_client: WorkspaceClient,
|
||||
workspace_client: Arc<dyn WorkspaceClient>,
|
||||
source: SourceRef,
|
||||
) -> Self {
|
||||
Self {
|
||||
@@ -615,7 +615,7 @@ mod tests {
|
||||
|
||||
fn stub_memory_backend_response(
|
||||
body: &'static str,
|
||||
) -> (WorkspaceClient, mpsc::Receiver<String>) {
|
||||
) -> (Arc<dyn WorkspaceClient>, mpsc::Receiver<String>) {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let (tx, rx) = mpsc::channel();
|
||||
@@ -659,7 +659,11 @@ mod tests {
|
||||
stream.write_all(response.as_bytes()).unwrap();
|
||||
});
|
||||
(
|
||||
WorkspaceClient::http("test-workspace", format!("http://{addr}")),
|
||||
Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new(
|
||||
"test-workspace",
|
||||
format!("http://{addr}"),
|
||||
"test-worker",
|
||||
)),
|
||||
rx,
|
||||
)
|
||||
}
|
||||
@@ -668,7 +672,7 @@ mod tests {
|
||||
fn descriptor_declares_session_explore_tools() {
|
||||
let state = SessionExploreState::new(
|
||||
SessionReferenceView::new("segment-1", vec![Item::user_message("remember this")]),
|
||||
WorkspaceClient::available("test-backend"),
|
||||
crate::worker::marker_workspace_client(None, "test-backend"),
|
||||
SourceRef {
|
||||
segment_id: "segment-1".to_string(),
|
||||
range: [0, 0],
|
||||
|
||||
@@ -4,7 +4,10 @@
|
||||
//! module only resolves the local backend root, declares the built-in feature,
|
||||
//! and contributes those tools through the normal feature registry path.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::{
|
||||
path::{Path, PathBuf},
|
||||
sync::Arc,
|
||||
};
|
||||
|
||||
use ticket::{
|
||||
LocalTicketBackend, MarkdownText, NewOrchestrationPlanRecord, NewTicket, NewTicketEvent,
|
||||
@@ -22,6 +25,7 @@ use crate::feature::{
|
||||
FeatureInstructionContribution, FeatureInstructionDeclaration, FeatureInstructionId,
|
||||
FeatureModule, ToolContribution, ToolDeclaration,
|
||||
};
|
||||
use crate::worker::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod};
|
||||
|
||||
const FEATURE_ID: &str = "ticket";
|
||||
const FEATURE_NAME: &str = "Ticket tools";
|
||||
@@ -183,13 +187,8 @@ const ORCHESTRATION_CONTROL_ADDITIONAL_TOOL_NAMES: &[&str] = &[
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub enum TicketFeatureBackend {
|
||||
Local {
|
||||
root: PathBuf,
|
||||
},
|
||||
WorkspaceHttp {
|
||||
workspace_id: String,
|
||||
base_url: String,
|
||||
},
|
||||
Local { root: PathBuf },
|
||||
WorkspaceClient(Arc<dyn WorkspaceClient>),
|
||||
}
|
||||
|
||||
impl From<PathBuf> for TicketFeatureBackend {
|
||||
@@ -274,7 +273,7 @@ impl TicketFeature {
|
||||
pub fn backend_root(&self) -> Option<&Path> {
|
||||
match &self.backend {
|
||||
TicketFeatureBackend::Local { root } => Some(root),
|
||||
TicketFeatureBackend::WorkspaceHttp { .. } => None,
|
||||
TicketFeatureBackend::WorkspaceClient(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -321,14 +320,8 @@ impl TicketFeature {
|
||||
.into(),
|
||||
)
|
||||
}
|
||||
TicketFeatureBackend::WorkspaceHttp {
|
||||
workspace_id,
|
||||
base_url,
|
||||
} => Some(
|
||||
TicketToolBackend::new(WorkspaceHttpTicketBackend::new(
|
||||
workspace_id.clone(),
|
||||
base_url.clone(),
|
||||
))
|
||||
TicketFeatureBackend::WorkspaceClient(client) => Some(
|
||||
TicketToolBackend::new(WorkspaceHttpTicketBackend::new(client.clone()))
|
||||
.with_record_language(self.record_language.as_deref()),
|
||||
),
|
||||
}
|
||||
@@ -386,22 +379,18 @@ impl FeatureModule for TicketFeature {
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct WorkspaceHttpTicketBackend {
|
||||
workspace_id: String,
|
||||
base_url: String,
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
}
|
||||
|
||||
impl WorkspaceHttpTicketBackend {
|
||||
fn new(workspace_id: String, base_url: String) -> Self {
|
||||
Self {
|
||||
workspace_id,
|
||||
base_url: base_url.trim_end_matches('/').to_string(),
|
||||
}
|
||||
fn new(client: Arc<dyn WorkspaceClient>) -> Self {
|
||||
Self { client }
|
||||
}
|
||||
|
||||
fn endpoint(&self) -> String {
|
||||
format!(
|
||||
"{}/api/w/{}/tickets/backend",
|
||||
self.base_url, self.workspace_id
|
||||
"/api/w/{}/tickets/backend",
|
||||
self.client.workspace_id().unwrap_or_default()
|
||||
)
|
||||
}
|
||||
|
||||
@@ -409,44 +398,44 @@ impl WorkspaceHttpTicketBackend {
|
||||
&self,
|
||||
operation: TicketBackendOperation,
|
||||
) -> TicketResult<TicketBackendOperationResult> {
|
||||
let client = self.client.clone();
|
||||
let endpoint = self.endpoint();
|
||||
if tokio::runtime::Handle::try_current().is_ok() {
|
||||
return std::thread::spawn(move || Self::invoke_http(endpoint, operation))
|
||||
return std::thread::spawn(move || Self::invoke_client(client, endpoint, operation))
|
||||
.join()
|
||||
.map_err(|_| {
|
||||
TicketError::Conflict("ticket backend request thread panicked".to_string())
|
||||
})?;
|
||||
}
|
||||
Self::invoke_http(endpoint, operation)
|
||||
Self::invoke_client(client, endpoint, operation)
|
||||
}
|
||||
|
||||
fn invoke_http(
|
||||
fn invoke_client(
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
endpoint: String,
|
||||
operation: TicketBackendOperation,
|
||||
) -> TicketResult<TicketBackendOperationResult> {
|
||||
let body = serde_json::to_string(&operation).map_err(|error| {
|
||||
TicketError::Conflict(format!("serialize ticket operation: {error}"))
|
||||
})?;
|
||||
let response = reqwest::blocking::Client::new()
|
||||
.post(endpoint)
|
||||
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
||||
.body(body)
|
||||
.send()
|
||||
let response = client
|
||||
.execute(WorkspaceRequest::json(
|
||||
WorkspaceRequestMethod::Post,
|
||||
endpoint,
|
||||
body,
|
||||
))
|
||||
.map_err(|error| {
|
||||
TicketError::Conflict(format!("ticket backend request failed: {error}"))
|
||||
})?;
|
||||
let status = response.status();
|
||||
let text = response.text().map_err(|error| {
|
||||
TicketError::Conflict(format!("ticket backend response failed: {error}"))
|
||||
})?;
|
||||
if !status.is_success() {
|
||||
if !response.is_success() {
|
||||
return Err(TicketError::Conflict(format!(
|
||||
"ticket backend returned HTTP {status}: {text}"
|
||||
"ticket backend returned HTTP {}: {}",
|
||||
response.status, response.body
|
||||
)));
|
||||
}
|
||||
match serde_json::from_str::<TicketBackendHttpResponse>(&text).map_err(|error| {
|
||||
TicketError::Conflict(format!("decode ticket backend response: {error}"))
|
||||
})? {
|
||||
match serde_json::from_str::<TicketBackendHttpResponse>(&response.body).map_err(
|
||||
|error| TicketError::Conflict(format!("decode ticket backend response: {error}")),
|
||||
)? {
|
||||
TicketBackendHttpResponse::Ok { result } => Ok(result),
|
||||
TicketBackendHttpResponse::Error { message } => Err(TicketError::Conflict(message)),
|
||||
}
|
||||
@@ -1126,8 +1115,13 @@ provider = "github"
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn workspace_http_backend_invoke_is_safe_inside_async_context() {
|
||||
let backend =
|
||||
WorkspaceHttpTicketBackend::new("workspace-a".to_string(), "not-a-url".to_string());
|
||||
let backend = WorkspaceHttpTicketBackend::new(Arc::new(
|
||||
crate::worker::RuntimeWorkspaceHttpClient::new(
|
||||
"workspace-a",
|
||||
"not-a-url",
|
||||
"test-worker",
|
||||
),
|
||||
));
|
||||
|
||||
let error = backend
|
||||
.invoke(TicketBackendOperation::DefaultIntakeReadyStateChangeBody {
|
||||
@@ -1167,7 +1161,9 @@ provider = "github"
|
||||
.unwrap();
|
||||
});
|
||||
|
||||
let backend = WorkspaceHttpTicketBackend::new("workspace-a".to_string(), base_url);
|
||||
let backend = WorkspaceHttpTicketBackend::new(Arc::new(
|
||||
crate::worker::RuntimeWorkspaceHttpClient::new("workspace-a", base_url, "test-worker"),
|
||||
));
|
||||
let created = backend.create(NewTicket::new("HTTP ticket")).unwrap();
|
||||
|
||||
server.join().unwrap();
|
||||
|
||||
@@ -40,6 +40,9 @@ pub use runtime::dir::RuntimeDir;
|
||||
pub use segment_log_sink::SegmentLogSink;
|
||||
pub use shared_state::WorkerSharedState;
|
||||
pub use worker::{
|
||||
LocalWorkingDirectory, Worker, WorkerError, WorkerFilesystemAuthority, WorkerRunResult,
|
||||
WorkerWorkspaceContext, WorkspaceClient, WorkspaceId, WorkspaceIdError, apply_worker_manifest,
|
||||
LocalWorkingDirectory, RuntimeWorkspaceHttpClient, Worker, WorkerError,
|
||||
WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext, WorkspaceClient,
|
||||
WorkspaceClientError, WorkspaceId, WorkspaceIdError, WorkspaceRequest, WorkspaceRequestMethod,
|
||||
WorkspaceResponse, apply_worker_manifest, marker_workspace_client,
|
||||
unavailable_workspace_client,
|
||||
};
|
||||
|
||||
+33
-26
@@ -128,7 +128,7 @@ pub enum SkillClientError {
|
||||
#[error("workspace client kind `{0}` does not expose direct Skill HTTP operations")]
|
||||
UnsupportedClient(String),
|
||||
#[error("Skill request failed: {0}")]
|
||||
Request(#[from] reqwest::Error),
|
||||
Request(#[from] crate::worker::WorkspaceClientError),
|
||||
#[error("Skill API response JSON is invalid: {0}")]
|
||||
Json(#[from] serde_json::Error),
|
||||
#[error("Skill API returned HTTP {status}: {body}")]
|
||||
@@ -140,7 +140,7 @@ pub enum SkillClientError {
|
||||
InvalidBaseUrl(String),
|
||||
}
|
||||
|
||||
impl WorkspaceClient {
|
||||
impl dyn WorkspaceClient + '_ {
|
||||
pub fn list_skills(&self) -> Result<SkillCatalogResponse, SkillClientError> {
|
||||
self.get_skill_json("skills")
|
||||
}
|
||||
@@ -157,29 +157,21 @@ impl WorkspaceClient {
|
||||
&self,
|
||||
path: &str,
|
||||
) -> Result<T, SkillClientError> {
|
||||
let Self::Http {
|
||||
workspace_id,
|
||||
base_url,
|
||||
} = self
|
||||
else {
|
||||
return match self {
|
||||
Self::Available { kind } => Err(SkillClientError::UnsupportedClient(kind.clone())),
|
||||
Self::Unavailable { reason } => Err(SkillClientError::Unavailable(reason.clone())),
|
||||
Self::Http { .. } => unreachable!(),
|
||||
};
|
||||
};
|
||||
if base_url.trim().is_empty() {
|
||||
return Err(SkillClientError::InvalidBaseUrl(base_url.clone()));
|
||||
let workspace_id = self
|
||||
.workspace_id()
|
||||
.ok_or_else(|| SkillClientError::UnsupportedClient(self.kind().to_string()))?;
|
||||
let response = self.execute(crate::worker::WorkspaceRequest::get(format!(
|
||||
"/api/w/{workspace_id}/{path}"
|
||||
)))?;
|
||||
let status = reqwest::StatusCode::from_u16(response.status)
|
||||
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
|
||||
if !response.is_success() {
|
||||
return Err(SkillClientError::Http {
|
||||
status,
|
||||
body: response.body,
|
||||
});
|
||||
}
|
||||
let base = base_url.trim_end_matches('/');
|
||||
let url = format!("{base}/api/w/{workspace_id}/{path}");
|
||||
let response = reqwest::blocking::Client::new().get(url).send()?;
|
||||
let status = response.status();
|
||||
let body = response.text()?;
|
||||
if !status.is_success() {
|
||||
return Err(SkillClientError::Http { status, body });
|
||||
}
|
||||
Ok(serde_json::from_str(&body)?)
|
||||
Ok(serde_json::from_str(&response.body)?)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -201,13 +193,23 @@ mod tests {
|
||||
let mut request_line = String::new();
|
||||
reader.read_line(&mut request_line).unwrap();
|
||||
assert!(request_line.starts_with("GET /api/w/ws-1/skills HTTP/1.1"));
|
||||
let mut worker_header = None;
|
||||
let mut authorization = None;
|
||||
loop {
|
||||
let mut line = String::new();
|
||||
reader.read_line(&mut line).unwrap();
|
||||
if let Some(value) = line.strip_prefix("x-yoi-worker-id: ") {
|
||||
worker_header = Some(value.trim().to_string());
|
||||
}
|
||||
if let Some(value) = line.strip_prefix("authorization: ") {
|
||||
authorization = Some(value.trim().to_string());
|
||||
}
|
||||
if line == "\r\n" || line.is_empty() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert_eq!(worker_header.as_deref(), Some("test-worker"));
|
||||
assert_eq!(authorization.as_deref(), Some("Bearer test-credential"));
|
||||
let body = serde_json::json!({
|
||||
"authority": "workspace-backend-skills-v0",
|
||||
"entries": [{
|
||||
@@ -229,8 +231,13 @@ mod tests {
|
||||
.unwrap();
|
||||
});
|
||||
|
||||
let client = WorkspaceClient::http("ws-1", format!("http://{addr}"));
|
||||
let catalog = client.list_skills().unwrap();
|
||||
let client = crate::worker::RuntimeWorkspaceHttpClient::new(
|
||||
"ws-1",
|
||||
format!("http://{addr}"),
|
||||
"test-worker",
|
||||
)
|
||||
.with_access_token(Some("test-credential".to_string()));
|
||||
let catalog = (&client as &dyn WorkspaceClient).list_skills().unwrap();
|
||||
assert_eq!(catalog.entries[0].name, "triage-errors");
|
||||
assert_eq!(catalog.entries[0].provenance.id, "workspace:triage-errors");
|
||||
handle.join().unwrap();
|
||||
|
||||
+428
-49
@@ -143,77 +143,368 @@ pub enum WorkspaceIdError {
|
||||
Empty,
|
||||
}
|
||||
|
||||
/// Narrow path-free workspace API handle injected by Runtime/host code.
|
||||
///
|
||||
/// This is deliberately not a filesystem authority surface. A Worker may have a
|
||||
/// workspace client without local filesystem authority, or neither. Local
|
||||
/// path-backed implementations are represented only as a capability marker here;
|
||||
/// the actual paths remain under [`WorkerFilesystemAuthority::Local`] or in host
|
||||
/// adapter code.
|
||||
/// One authority-bound operation sent through the Runtime-supplied Workspace client.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum WorkspaceClient {
|
||||
/// Runtime/host supplied an HTTP workspace API endpoint.
|
||||
Http {
|
||||
workspace_id: String,
|
||||
base_url: String,
|
||||
},
|
||||
/// Runtime/host supplied a workspace API handle. The string is an opaque
|
||||
/// diagnostic/backend kind, not an endpoint, path, or secret-bearing value.
|
||||
Available { kind: String },
|
||||
/// Workspace-aware operations must fail closed or stay disabled.
|
||||
Unavailable { reason: String },
|
||||
pub struct WorkspaceRequest {
|
||||
pub method: WorkspaceRequestMethod,
|
||||
pub path: String,
|
||||
pub body: Option<String>,
|
||||
}
|
||||
|
||||
impl WorkspaceClient {
|
||||
pub fn available(kind: impl Into<String>) -> Self {
|
||||
Self::Available { kind: kind.into() }
|
||||
impl WorkspaceRequest {
|
||||
pub fn get(path: impl Into<String>) -> Self {
|
||||
Self {
|
||||
method: WorkspaceRequestMethod::Get,
|
||||
path: path.into(),
|
||||
body: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn http(workspace_id: impl Into<String>, base_url: impl Into<String>) -> Self {
|
||||
Self::Http {
|
||||
pub fn json(
|
||||
method: WorkspaceRequestMethod,
|
||||
path: impl Into<String>,
|
||||
body: impl Into<String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
method,
|
||||
path: path.into(),
|
||||
body: Some(body.into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum WorkspaceRequestMethod {
|
||||
Get,
|
||||
Post,
|
||||
Put,
|
||||
Patch,
|
||||
Delete,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct WorkspaceResponse {
|
||||
pub status: u16,
|
||||
pub body: String,
|
||||
}
|
||||
|
||||
impl WorkspaceResponse {
|
||||
pub fn is_success(&self) -> bool {
|
||||
(200..300).contains(&self.status)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum WorkspaceClientError {
|
||||
#[error("workspace client is unavailable: {0}")]
|
||||
Unavailable(String),
|
||||
#[error("workspace request path must start with '/': {0}")]
|
||||
InvalidPath(String),
|
||||
#[error("workspace request failed: {0}")]
|
||||
Request(String),
|
||||
}
|
||||
|
||||
/// Path-free Workspace operation authority injected by Runtime/host code.
|
||||
///
|
||||
/// Workers receive this trait object rather than a Backend URL. The concrete
|
||||
/// implementation is responsible for binding Runtime/Worker identity and
|
||||
/// forwarding operations to the Workspace authority.
|
||||
pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
|
||||
fn workspace_id(&self) -> Option<&str>;
|
||||
fn kind(&self) -> &str;
|
||||
fn is_available(&self) -> bool;
|
||||
fn execute(&self, request: WorkspaceRequest)
|
||||
-> Result<WorkspaceResponse, WorkspaceClientError>;
|
||||
}
|
||||
|
||||
/// HTTP forwarding client created by Runtime for one concrete Worker execution.
|
||||
///
|
||||
/// The upstream endpoint and source headers are private implementation details;
|
||||
/// model-visible tools can only submit [`WorkspaceRequest`] values through the
|
||||
/// [`WorkspaceClient`] trait.
|
||||
pub struct RuntimeWorkspaceHttpClient {
|
||||
workspace_id: String,
|
||||
base_url: String,
|
||||
worker_id: String,
|
||||
access_token: Mutex<Option<String>>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for RuntimeWorkspaceHttpClient {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
formatter
|
||||
.debug_struct("RuntimeWorkspaceHttpClient")
|
||||
.field("workspace_id", &self.workspace_id)
|
||||
.field("base_url", &self.base_url)
|
||||
.field("worker_id", &self.worker_id)
|
||||
.field(
|
||||
"access_token",
|
||||
&self
|
||||
.access_token
|
||||
.lock()
|
||||
.ok()
|
||||
.and_then(|token| token.as_ref().map(|_| "[redacted]")),
|
||||
)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl RuntimeWorkspaceHttpClient {
|
||||
pub fn new(
|
||||
workspace_id: impl Into<String>,
|
||||
base_url: impl Into<String>,
|
||||
worker_id: impl Into<String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
workspace_id: workspace_id.into(),
|
||||
base_url: base_url.into(),
|
||||
base_url: base_url.into().trim_end_matches('/').to_string(),
|
||||
worker_id: worker_id.into(),
|
||||
access_token: Mutex::new(None),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn unavailable(reason: impl Into<String>) -> Self {
|
||||
Self::Unavailable {
|
||||
pub fn with_access_token(self, access_token: Option<String>) -> Self {
|
||||
*self.access_token.lock().expect("new credential mutex") = access_token;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkspaceClient for RuntimeWorkspaceHttpClient {
|
||||
fn workspace_id(&self) -> Option<&str> {
|
||||
Some(&self.workspace_id)
|
||||
}
|
||||
|
||||
fn kind(&self) -> &str {
|
||||
"runtime-http-proxy"
|
||||
}
|
||||
|
||||
fn is_available(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn execute(
|
||||
&self,
|
||||
request: WorkspaceRequest,
|
||||
) -> Result<WorkspaceResponse, WorkspaceClientError> {
|
||||
let base_url = self.base_url.clone();
|
||||
let worker_id = self.worker_id.clone();
|
||||
let access_token = self
|
||||
.access_token
|
||||
.lock()
|
||||
.map_err(|_| {
|
||||
WorkspaceClientError::Request("workspace credential lock poisoned".to_string())
|
||||
})?
|
||||
.clone();
|
||||
let request_copy = request.clone();
|
||||
let result = if tokio::runtime::Handle::try_current().is_ok() {
|
||||
std::thread::spawn(move || {
|
||||
execute_runtime_workspace_http_with_refresh(
|
||||
&base_url,
|
||||
&worker_id,
|
||||
access_token,
|
||||
request_copy,
|
||||
)
|
||||
})
|
||||
.join()
|
||||
.map_err(|_| {
|
||||
WorkspaceClientError::Request("workspace request thread panicked".to_string())
|
||||
})?
|
||||
} else {
|
||||
execute_runtime_workspace_http_with_refresh(
|
||||
&base_url,
|
||||
&worker_id,
|
||||
access_token,
|
||||
request,
|
||||
)
|
||||
}?;
|
||||
if let Some(new_token) = result.1 {
|
||||
*self.access_token.lock().map_err(|_| {
|
||||
WorkspaceClientError::Request("workspace credential lock poisoned".to_string())
|
||||
})? = Some(new_token);
|
||||
}
|
||||
Ok(result.0)
|
||||
}
|
||||
}
|
||||
|
||||
fn execute_runtime_workspace_http_with_refresh(
|
||||
base_url: &str,
|
||||
worker_id: &str,
|
||||
access_token: Option<String>,
|
||||
request: WorkspaceRequest,
|
||||
) -> Result<(WorkspaceResponse, Option<String>), WorkspaceClientError> {
|
||||
let response = execute_runtime_workspace_http(
|
||||
base_url,
|
||||
worker_id,
|
||||
access_token.as_deref(),
|
||||
request.clone(),
|
||||
)?;
|
||||
if response.status != 401 {
|
||||
return Ok((response, None));
|
||||
}
|
||||
let Some(expired_token) = access_token else {
|
||||
return Ok((response, None));
|
||||
};
|
||||
let workspace_id = request
|
||||
.path
|
||||
.strip_prefix("/api/w/")
|
||||
.and_then(|path| path.split('/').next())
|
||||
.ok_or_else(|| WorkspaceClientError::InvalidPath(request.path.clone()))?;
|
||||
let refresh_url = format!("{base_url}/api/w/{workspace_id}/worker-credentials/refresh");
|
||||
let refresh = reqwest::blocking::Client::new()
|
||||
.post(refresh_url)
|
||||
.bearer_auth(expired_token)
|
||||
.header("x-yoi-worker-id", worker_id)
|
||||
.send()
|
||||
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
|
||||
if !refresh.status().is_success() {
|
||||
return Ok((response, None));
|
||||
}
|
||||
let body: serde_json::Value = refresh
|
||||
.json()
|
||||
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
|
||||
let new_token = body
|
||||
.get("access_token")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.ok_or_else(|| {
|
||||
WorkspaceClientError::Request(
|
||||
"Workspace credential refresh response omitted access_token".to_string(),
|
||||
)
|
||||
})?
|
||||
.to_string();
|
||||
let retried = execute_runtime_workspace_http(base_url, worker_id, Some(&new_token), request)?;
|
||||
Ok((retried, Some(new_token)))
|
||||
}
|
||||
|
||||
fn execute_runtime_workspace_http(
|
||||
base_url: &str,
|
||||
worker_id: &str,
|
||||
access_token: Option<&str>,
|
||||
request: WorkspaceRequest,
|
||||
) -> Result<WorkspaceResponse, WorkspaceClientError> {
|
||||
if !request.path.starts_with('/') || request.path.starts_with("//") {
|
||||
return Err(WorkspaceClientError::InvalidPath(request.path));
|
||||
}
|
||||
let url = format!("{base_url}{}", request.path);
|
||||
let method = match request.method {
|
||||
WorkspaceRequestMethod::Get => reqwest::Method::GET,
|
||||
WorkspaceRequestMethod::Post => reqwest::Method::POST,
|
||||
WorkspaceRequestMethod::Put => reqwest::Method::PUT,
|
||||
WorkspaceRequestMethod::Patch => reqwest::Method::PATCH,
|
||||
WorkspaceRequestMethod::Delete => reqwest::Method::DELETE,
|
||||
};
|
||||
let client = reqwest::blocking::Client::new();
|
||||
let mut request_builder = client
|
||||
.request(method, url)
|
||||
.header("x-yoi-worker-id", worker_id);
|
||||
if let Some(access_token) = access_token {
|
||||
request_builder = request_builder.bearer_auth(access_token);
|
||||
}
|
||||
if let Some(body) = request.body {
|
||||
request_builder = request_builder
|
||||
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
||||
.body(body);
|
||||
}
|
||||
let response = request_builder
|
||||
.send()
|
||||
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
|
||||
let status = response.status().as_u16();
|
||||
let body = response
|
||||
.text()
|
||||
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
|
||||
Ok(WorkspaceResponse { status, body })
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct MarkerWorkspaceClient {
|
||||
workspace_id: Option<String>,
|
||||
kind: String,
|
||||
available: bool,
|
||||
reason: String,
|
||||
}
|
||||
|
||||
impl WorkspaceClient for MarkerWorkspaceClient {
|
||||
fn workspace_id(&self) -> Option<&str> {
|
||||
self.workspace_id.as_deref()
|
||||
}
|
||||
|
||||
fn kind(&self) -> &str {
|
||||
&self.kind
|
||||
}
|
||||
|
||||
fn is_available(&self) -> bool {
|
||||
self.available
|
||||
}
|
||||
|
||||
fn execute(
|
||||
&self,
|
||||
_request: WorkspaceRequest,
|
||||
) -> Result<WorkspaceResponse, WorkspaceClientError> {
|
||||
Err(WorkspaceClientError::Unavailable(self.reason.clone()))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn unavailable_workspace_client(
|
||||
workspace_id: Option<&WorkspaceId>,
|
||||
reason: impl Into<String>,
|
||||
) -> Arc<dyn WorkspaceClient> {
|
||||
Arc::new(MarkerWorkspaceClient {
|
||||
workspace_id: workspace_id.map(|id| id.as_str().to_string()),
|
||||
kind: "unavailable".to_string(),
|
||||
available: false,
|
||||
reason: reason.into(),
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
pub fn local_filesystem() -> Self {
|
||||
Self::available("local-filesystem")
|
||||
}
|
||||
|
||||
pub fn is_available(&self) -> bool {
|
||||
matches!(self, Self::Available { .. } | Self::Http { .. })
|
||||
}
|
||||
pub fn marker_workspace_client(
|
||||
workspace_id: Option<&WorkspaceId>,
|
||||
kind: impl Into<String>,
|
||||
) -> Arc<dyn WorkspaceClient> {
|
||||
let kind = kind.into();
|
||||
Arc::new(MarkerWorkspaceClient {
|
||||
workspace_id: workspace_id.map(|id| id.as_str().to_string()),
|
||||
reason: format!("workspace client kind `{kind}` does not expose Workspace operations"),
|
||||
kind,
|
||||
available: true,
|
||||
})
|
||||
}
|
||||
|
||||
/// Workspace context supplied to a Worker separately from filesystem authority.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
#[derive(Clone)]
|
||||
pub struct WorkerWorkspaceContext {
|
||||
workspace_id: Option<WorkspaceId>,
|
||||
client: WorkspaceClient,
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for WorkerWorkspaceContext {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
formatter
|
||||
.debug_struct("WorkerWorkspaceContext")
|
||||
.field("workspace_id", &self.workspace_id)
|
||||
.field("client_kind", &self.client.kind())
|
||||
.field("client_available", &self.client.is_available())
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkerWorkspaceContext {
|
||||
pub fn no_workspace() -> Self {
|
||||
Self {
|
||||
workspace_id: None,
|
||||
client: WorkspaceClient::unavailable("no workspace configured"),
|
||||
client: unavailable_workspace_client(None, "no workspace configured"),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn unavailable(workspace_id: Option<WorkspaceId>, reason: impl Into<String>) -> Self {
|
||||
let client = unavailable_workspace_client(workspace_id.as_ref(), reason);
|
||||
Self {
|
||||
workspace_id,
|
||||
client: WorkspaceClient::unavailable(reason),
|
||||
client,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_client(workspace_id: Option<WorkspaceId>, client: WorkspaceClient) -> Self {
|
||||
pub fn with_client(
|
||||
workspace_id: Option<WorkspaceId>,
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
) -> Self {
|
||||
Self {
|
||||
workspace_id,
|
||||
client,
|
||||
@@ -221,15 +512,23 @@ impl WorkerWorkspaceContext {
|
||||
}
|
||||
|
||||
pub fn local_filesystem(workspace_id: Option<WorkspaceId>) -> Self {
|
||||
Self::with_client(workspace_id, WorkspaceClient::local_filesystem())
|
||||
let client = marker_workspace_client(workspace_id.as_ref(), "local-filesystem");
|
||||
Self {
|
||||
workspace_id,
|
||||
client,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn workspace_id(&self) -> Option<&WorkspaceId> {
|
||||
self.workspace_id.as_ref()
|
||||
}
|
||||
|
||||
pub fn client(&self) -> &WorkspaceClient {
|
||||
&self.client
|
||||
pub fn client(&self) -> &dyn WorkspaceClient {
|
||||
self.client.as_ref()
|
||||
}
|
||||
|
||||
pub fn client_handle(&self) -> Arc<dyn WorkspaceClient> {
|
||||
self.client.clone()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -926,10 +1225,14 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
|
||||
|
||||
/// Narrow workspace client/availability handle injected by Runtime/host.
|
||||
/// This never grants local filesystem authority.
|
||||
pub fn workspace_client(&self) -> &WorkspaceClient {
|
||||
pub fn workspace_client(&self) -> &dyn WorkspaceClient {
|
||||
self.workspace_context.client()
|
||||
}
|
||||
|
||||
pub fn workspace_client_handle(&self) -> Arc<dyn WorkspaceClient> {
|
||||
self.workspace_context.client_handle()
|
||||
}
|
||||
|
||||
async fn resident_summary_from_workspace_authority(
|
||||
&self,
|
||||
) -> Result<Option<String>, WorkerError> {
|
||||
@@ -3197,7 +3500,7 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
|
||||
items_to_extract,
|
||||
);
|
||||
let session_explore_state =
|
||||
SessionExploreState::new(session_view, self.workspace_client().clone(), source);
|
||||
SessionExploreState::new(session_view, self.workspace_client_handle(), source);
|
||||
let input_text = render_extract_input(session_explore_state.view());
|
||||
let mut internal_tools = Vec::new();
|
||||
let mut internal_hook_builder = HookRegistryBuilder::new();
|
||||
@@ -3464,7 +3767,7 @@ impl WorkerAuditBase {
|
||||
|
||||
async fn emit(
|
||||
&self,
|
||||
workspace_client: &WorkspaceClient,
|
||||
workspace_client: &dyn WorkspaceClient,
|
||||
event_tx: Option<&broadcast::Sender<Event>>,
|
||||
status: memory::audit::WorkerLifecycleStatus,
|
||||
reason: impl Into<String>,
|
||||
@@ -4936,7 +5239,7 @@ mod spawned_context_tests {
|
||||
false,
|
||||
WorkerWorkspaceContext::with_client(
|
||||
Some(workspace_id.clone()),
|
||||
WorkspaceClient::available("test-api"),
|
||||
marker_workspace_client(Some(&workspace_id), "test-api"),
|
||||
),
|
||||
WorkerFilesystemAuthority::None,
|
||||
manifest.scope.clone(),
|
||||
@@ -5773,7 +6076,11 @@ mod build_summary_prompt_tests {
|
||||
});
|
||||
WorkerWorkspaceContext::with_client(
|
||||
Some(WorkspaceId::new("test-memory").unwrap()),
|
||||
WorkspaceClient::http("test-memory", format!("http://{addr}")),
|
||||
Arc::new(RuntimeWorkspaceHttpClient::new(
|
||||
"test-memory",
|
||||
format!("http://{addr}"),
|
||||
"test-worker",
|
||||
)),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -5905,7 +6212,11 @@ mod build_summary_prompt_tests {
|
||||
store,
|
||||
WorkerWorkspaceContext::with_client(
|
||||
Some(WorkspaceId::new("ws-skill").unwrap()),
|
||||
WorkspaceClient::http("ws-skill", format!("http://{addr}")),
|
||||
Arc::new(RuntimeWorkspaceHttpClient::new(
|
||||
"ws-skill",
|
||||
format!("http://{addr}"),
|
||||
"test-worker",
|
||||
)),
|
||||
),
|
||||
authority,
|
||||
scope,
|
||||
@@ -5946,6 +6257,74 @@ mod build_summary_prompt_tests {
|
||||
}));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runtime_workspace_client_refreshes_expired_credential_and_retries() {
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
use std::net::TcpListener;
|
||||
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let address = listener.local_addr().unwrap();
|
||||
let server = std::thread::spawn(move || {
|
||||
for step in 0..3 {
|
||||
let (mut stream, _) = listener.accept().unwrap();
|
||||
let mut reader = BufReader::new(stream.try_clone().unwrap());
|
||||
let mut first_line = String::new();
|
||||
reader.read_line(&mut first_line).unwrap();
|
||||
let mut authorization = String::new();
|
||||
loop {
|
||||
let mut line = String::new();
|
||||
reader.read_line(&mut line).unwrap();
|
||||
if let Some(value) = line.strip_prefix("authorization: ") {
|
||||
authorization = value.trim().to_string();
|
||||
}
|
||||
if line == "\r\n" || line.is_empty() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
match step {
|
||||
0 => {
|
||||
assert_eq!(authorization, "Bearer expired-token");
|
||||
stream
|
||||
.write_all(b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n")
|
||||
.unwrap();
|
||||
}
|
||||
1 => {
|
||||
assert!(first_line.contains("/worker-credentials/refresh"));
|
||||
assert_eq!(authorization, "Bearer expired-token");
|
||||
let body =
|
||||
r#"{"access_token":"fresh-token","expires_at":"2099-01-01T00:00:00Z"}"#;
|
||||
write!(
|
||||
stream,
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}",
|
||||
body.len(),
|
||||
body
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
_ => {
|
||||
assert_eq!(authorization, "Bearer fresh-token");
|
||||
stream
|
||||
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}")
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
let client = RuntimeWorkspaceHttpClient::new(
|
||||
"workspace-refresh",
|
||||
format!("http://{address}"),
|
||||
"worker-refresh",
|
||||
)
|
||||
.with_access_token(Some("expired-token".to_string()));
|
||||
let response = client
|
||||
.execute(WorkspaceRequest::get(
|
||||
"/api/w/workspace-refresh/tickets/backend",
|
||||
))
|
||||
.unwrap();
|
||||
assert_eq!(response.status, 200);
|
||||
server.join().unwrap();
|
||||
}
|
||||
|
||||
fn minimal_manifest() -> WorkerManifest {
|
||||
let toml_str = r#"
|
||||
[worker]
|
||||
|
||||
@@ -308,6 +308,27 @@ pub struct WorkerSpawnWorkingDirectoryRequest {
|
||||
pub selector: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct WorkerTicketAssignmentRequest {
|
||||
pub ticket_id: String,
|
||||
pub operation_id: String,
|
||||
}
|
||||
|
||||
pub(crate) fn worker_spawn_idempotency(
|
||||
request: &WorkerSpawnRequest,
|
||||
) -> Result<Option<(String, String)>, String> {
|
||||
let Some(assignment) = request.ticket_assignment.as_ref() else {
|
||||
return Ok(None);
|
||||
};
|
||||
let encoded = serde_json::to_vec(request)
|
||||
.map_err(|error| format!("serialize Worker spawn idempotency input: {error}"))?;
|
||||
Ok(Some((
|
||||
assignment.operation_id.clone(),
|
||||
format!("sha256:{}", digest_hex(&encoded, 64)),
|
||||
)))
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct WorkerSpawnRequest {
|
||||
@@ -317,6 +338,8 @@ pub struct WorkerSpawnRequest {
|
||||
pub acceptance: WorkerSpawnAcceptanceRequirement,
|
||||
pub profile: ProfileSelector,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub ticket_assignment: Option<WorkerTicketAssignmentRequest>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub initial_input: Option<EmbeddedWorkerInput>,
|
||||
/// Optional safe working-directory creation request. The Workspace server resolves
|
||||
/// this into a runtime-internal `WorkingDirectoryRequest` from configured
|
||||
@@ -329,6 +352,8 @@ pub struct WorkerSpawnRequest {
|
||||
pub resolved_working_directory: Option<WorkingDirectoryClaim>,
|
||||
#[serde(skip, default)]
|
||||
pub resolved_config_bundle: Option<ConfigBundle>,
|
||||
#[serde(skip, default)]
|
||||
pub resolved_workspace_api: Option<WorkspaceApiRef>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
@@ -427,6 +452,8 @@ pub struct WorkerStopResult {
|
||||
pub struct WorkerLifecycleRequest {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub reason: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub ticket_assignment: Option<WorkerTicketAssignmentRequest>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
@@ -1695,7 +1722,14 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
|
||||
};
|
||||
}
|
||||
};
|
||||
let (idempotency_key, idempotency_fingerprint) = worker_spawn_idempotency(&request)
|
||||
.expect("WorkerSpawnRequest serialization is infallible")
|
||||
.map_or((None, None), |(key, fingerprint)| {
|
||||
(Some(key), Some(fingerprint))
|
||||
});
|
||||
let create_request = CreateWorkerRequest {
|
||||
idempotency_key,
|
||||
idempotency_fingerprint,
|
||||
profile,
|
||||
display_name: request.requested_worker_name.clone(),
|
||||
config_bundle: None,
|
||||
@@ -1703,12 +1737,15 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
|
||||
initial_input: request.initial_input.clone(),
|
||||
working_directory_request: request.resolved_working_directory_request.clone(),
|
||||
working_directory: request.resolved_working_directory.clone(),
|
||||
workspace_api: self
|
||||
.backend_base_url
|
||||
workspace_api: request.resolved_workspace_api.clone().or_else(|| {
|
||||
self.backend_base_url
|
||||
.as_ref()
|
||||
.map(|base_url| WorkspaceApiRef {
|
||||
workspace_id: self.workspace_id.clone(),
|
||||
base_url: base_url.clone(),
|
||||
runtime_id: Some(self.runtime_id.clone()),
|
||||
access_token: None,
|
||||
})
|
||||
}),
|
||||
};
|
||||
match self.runtime.create_worker(create_request) {
|
||||
@@ -2669,7 +2706,14 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
|
||||
};
|
||||
}
|
||||
};
|
||||
let (idempotency_key, idempotency_fingerprint) = worker_spawn_idempotency(&request)
|
||||
.expect("WorkerSpawnRequest serialization is infallible")
|
||||
.map_or((None, None), |(key, fingerprint)| {
|
||||
(Some(key), Some(fingerprint))
|
||||
});
|
||||
let create = CreateWorkerRequest {
|
||||
idempotency_key,
|
||||
idempotency_fingerprint,
|
||||
profile,
|
||||
display_name: request.requested_worker_name.clone(),
|
||||
config_bundle: None,
|
||||
@@ -2677,9 +2721,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
|
||||
initial_input: request.initial_input.clone(),
|
||||
working_directory_request: request.resolved_working_directory_request.clone(),
|
||||
working_directory: request.resolved_working_directory.clone(),
|
||||
workspace_api: Some(WorkspaceApiRef {
|
||||
workspace_api: request.resolved_workspace_api.clone().or_else(|| {
|
||||
Some(WorkspaceApiRef {
|
||||
workspace_id: self.workspace_id.clone(),
|
||||
base_url: self.backend_base_url.clone(),
|
||||
runtime_id: Some(self.runtime_id.clone()),
|
||||
access_token: None,
|
||||
})
|
||||
}),
|
||||
};
|
||||
match self.post_json::<_, RuntimeHttpWorkerResponse>("/v1/workers", &create) {
|
||||
@@ -3121,8 +3169,11 @@ fn embedded_profile_path(profile: &ProfileSelector) -> Result<String, String> {
|
||||
fn embedded_profile_label(profile: &ProfileSelector) -> Option<String> {
|
||||
Some(match profile {
|
||||
ProfileSelector::Builtin(name) | ProfileSelector::Named(name) => {
|
||||
if name.strip_prefix("builtin:").unwrap_or(name) == MEMORY_CONSOLIDATION_PROFILE {
|
||||
let builtin_name = name.strip_prefix("builtin:").unwrap_or(name);
|
||||
if builtin_name == MEMORY_CONSOLIDATION_PROFILE {
|
||||
MEMORY_CONSOLIDATION_PROFILE.to_string()
|
||||
} else if builtin_name == WORKSPACE_ORCHESTRATOR_PROFILE {
|
||||
WORKSPACE_ORCHESTRATOR_PROFILE.to_string()
|
||||
} else {
|
||||
safe_display_hint(name)
|
||||
}
|
||||
@@ -3132,6 +3183,8 @@ fn embedded_profile_label(profile: &ProfileSelector) -> Option<String> {
|
||||
|
||||
const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation";
|
||||
const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation";
|
||||
const WORKSPACE_ORCHESTRATOR_PROFILE: &str = "orchestrator";
|
||||
pub(crate) const WORKSPACE_ORCHESTRATOR_SINGLETON_KEY: &str = "workspace-orchestrator";
|
||||
|
||||
struct WorkerDisplayMetadata {
|
||||
display_name: String,
|
||||
@@ -3160,6 +3213,20 @@ fn worker_display_metadata(
|
||||
tags,
|
||||
};
|
||||
}
|
||||
if profile_label == Some(WORKSPACE_ORCHESTRATOR_PROFILE) {
|
||||
let mut tags = vec!["orchestrator".to_string(), "singleton".to_string()];
|
||||
if internal {
|
||||
tags.insert(0, "internal".to_string());
|
||||
}
|
||||
return WorkerDisplayMetadata {
|
||||
display_name: requested_display_name
|
||||
.filter(|value| !value.trim().is_empty())
|
||||
.map(safe_display_hint)
|
||||
.unwrap_or_else(|| "Workspace Orchestrator".to_string()),
|
||||
singleton_key: Some(WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string()),
|
||||
tags,
|
||||
};
|
||||
}
|
||||
let display_name = requested_display_name
|
||||
.filter(|value| !value.trim().is_empty())
|
||||
.map(safe_display_hint)
|
||||
@@ -4149,11 +4216,13 @@ mod tests {
|
||||
expected_segments: 0,
|
||||
},
|
||||
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
|
||||
ticket_assignment: None,
|
||||
initial_input: None,
|
||||
working_directory_request: None,
|
||||
resolved_working_directory_request: None,
|
||||
resolved_working_directory: None,
|
||||
resolved_config_bundle: None,
|
||||
resolved_workspace_api: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4275,11 +4344,13 @@ mod tests {
|
||||
expected_segments: 0,
|
||||
},
|
||||
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
|
||||
ticket_assignment: None,
|
||||
initial_input: None,
|
||||
working_directory_request: None,
|
||||
resolved_working_directory_request: None,
|
||||
resolved_working_directory: None,
|
||||
resolved_config_bundle: None,
|
||||
resolved_workspace_api: None,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
@@ -4371,11 +4442,13 @@ mod tests {
|
||||
expected_segments: 0,
|
||||
},
|
||||
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
|
||||
ticket_assignment: None,
|
||||
initial_input: None,
|
||||
working_directory_request: None,
|
||||
resolved_working_directory_request: None,
|
||||
resolved_working_directory: None,
|
||||
resolved_config_bundle: None,
|
||||
resolved_workspace_api: None,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
@@ -4403,11 +4476,13 @@ mod tests {
|
||||
requested_worker_name: None,
|
||||
acceptance: WorkerSpawnAcceptanceRequirement::SocketReady,
|
||||
profile: ProfileSelector::Builtin("builtin:companion".to_string()),
|
||||
ticket_assignment: None,
|
||||
initial_input: None,
|
||||
working_directory_request: None,
|
||||
resolved_working_directory_request: None,
|
||||
resolved_working_directory: None,
|
||||
resolved_config_bundle: None,
|
||||
resolved_workspace_api: None,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
@@ -85,6 +85,10 @@ pub enum Error {
|
||||
UnknownRepository(String),
|
||||
#[error("workspace id does not match this Workspace backend")]
|
||||
WorkspaceIdMismatch,
|
||||
#[error("Ticket assignment conflict: {0}")]
|
||||
TicketAssignmentConflict(String),
|
||||
#[error("Worker Workspace authentication failed: {0}")]
|
||||
WorkerWorkspaceAuthentication(String),
|
||||
#[error("workspace identity error: {0}")]
|
||||
WorkspaceIdentity(String),
|
||||
#[error("store error: {0}")]
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user