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

refactor: remove workflow machinery

Browse Source
This commit is contained in:
Hare
2026-07-16 05:45:24 +09:00
parent 83ad7506d7
commit d801b2698b
64 changed files with 141 additions and 4055 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/Cargo.toml
-1
View File
@@ -34,7 +34,6 @@ libc = { workspace = true }
schemars = { workspace = true }
ticket = { workspace = true }
memory = { workspace = true }
workflow-crate = { package = "workflow", path = "../workflow" }
uuid = { workspace = true, features = ["v7"] }
session-metrics = { workspace = true }
arc-swap = "1.9.1"
crates/worker/src/active_workflow.rs
-736
View File
@@ -1,736 +0,0 @@
//! Durable active workflow invocation state.
//!
//! Workflow bodies are resolved at invocation time and snapshotted here. The
//! snapshot, not whatever resource version is installed later, is the procedural
//! authority that survives compaction for the currently governed task.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use llm_engine::Item;
use llm_engine::tool::{
Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput,
};
use serde::{Deserialize, Serialize};
use serde_json::json;
use session_store::{LogEntry, SystemItem, segment_log};
pub const DOMAIN: &str = "worker.active_workflows";
pub const REHYDRATION_MESSAGE_PREFIX: &str = "[Active workflow snapshot]";
pub const INACTIVE_MESSAGE_PREFIX: &str = "[Active workflow state]";
const SCHEMA_VERSION: u32 = 1;
pub type LogEntryCommitter = Arc<dyn Fn(LogEntry) + Send + Sync>;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ActiveWorkflowSnapshot {
pub schema_version: u32,
pub workflows: Vec<ActiveWorkflowRecord>,
}
impl Default for ActiveWorkflowSnapshot {
fn default() -> Self {
Self {
schema_version: SCHEMA_VERSION,
workflows: Vec::new(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ActiveWorkflowRecord {
pub slug: String,
pub status: ActiveWorkflowStatus,
pub invocation: WorkflowInvocationInfo,
pub task_scope: String,
pub body_snapshot_policy: WorkflowBodySnapshotPolicy,
pub guidance_snapshot: String,
pub obligations: Vec<String>,
pub checkpoints: Vec<WorkflowCheckpoint>,
pub updated_at_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub completion: Option<WorkflowCompletionInfo>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ActiveWorkflowStatus {
Active,
Completed,
Cancelled,
}
impl std::fmt::Display for ActiveWorkflowStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let s = match self {
Self::Active => "active",
Self::Completed => "completed",
Self::Cancelled => "cancelled",
};
f.write_str(s)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkflowInvocationInfo {
pub source: WorkflowInvocationSource,
pub invoked_at_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowInvocationSource {
UserWorkflowInvokeSegment,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowBodySnapshotPolicy {
SnapshottedAtInvocation,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkflowCheckpoint {
pub label: String,
pub status: WorkflowCheckpointStatus,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowCheckpointStatus {
Open,
Done,
Cancelled,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkflowCompletionInfo {
pub completed_at_ms: u64,
pub reason: String,
}
#[derive(Debug, Clone, Default)]
pub struct ActiveWorkflowStore {
inner: Arc<Mutex<ActiveWorkflowSnapshot>>,
}
impl ActiveWorkflowStore {
pub fn new() -> Self {
Self::default()
}
pub fn snapshot(&self) -> ActiveWorkflowSnapshot {
self.inner.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
pub fn replace_with(&self, snapshot: ActiveWorkflowSnapshot) {
*self.inner.lock().unwrap_or_else(|e| e.into_inner()) = snapshot;
}
pub fn active_records(&self) -> Vec<ActiveWorkflowRecord> {
self.snapshot()
.workflows
.into_iter()
.filter(|record| record.status == ActiveWorkflowStatus::Active)
.collect()
}
pub fn activate_from_system_items(
&self,
items: &[SystemItem],
task_scope: String,
invoked_at_ms: u64,
) -> bool {
let mut grouped: BTreeMap<String, Vec<String>> = BTreeMap::new();
for item in items {
if let SystemItem::Workflow { slug, body } = item {
grouped.entry(slug.clone()).or_default().push(body.clone());
}
}
if grouped.is_empty() {
return false;
}
let mut snapshot = self.snapshot();
snapshot.schema_version = SCHEMA_VERSION;
for (slug, bodies) in grouped {
let guidance_snapshot = bodies.join("\n\n---\n\n");
let obligations = extract_obligations(&guidance_snapshot);
let checkpoints = obligations
.iter()
.take(32)
.map(|label| WorkflowCheckpoint {
label: label.clone(),
status: WorkflowCheckpointStatus::Open,
})
.collect();
let record = ActiveWorkflowRecord {
slug: slug.clone(),
status: ActiveWorkflowStatus::Active,
invocation: WorkflowInvocationInfo {
source: WorkflowInvocationSource::UserWorkflowInvokeSegment,
invoked_at_ms,
},
task_scope: truncate_chars(&task_scope, 2_000),
body_snapshot_policy: WorkflowBodySnapshotPolicy::SnapshottedAtInvocation,
guidance_snapshot,
obligations,
checkpoints,
updated_at_ms: invoked_at_ms,
completion: None,
};
upsert_record(&mut snapshot.workflows, record);
}
self.replace_with(snapshot);
true
}
pub fn set_status(
&self,
slug: &str,
status: ActiveWorkflowStatus,
reason: String,
now_ms: u64,
) -> Result<ActiveWorkflowRecord, String> {
let mut snapshot = self.inner.lock().unwrap_or_else(|e| e.into_inner());
let record = snapshot
.workflows
.iter_mut()
.find(|record| record.slug == slug)
.ok_or_else(|| format!("active workflow `{slug}` not found"))?;
record.status = status;
record.updated_at_ms = now_ms;
record.completion = Some(WorkflowCompletionInfo {
completed_at_ms: now_ms,
reason,
});
for checkpoint in &mut record.checkpoints {
checkpoint.status = match status {
ActiveWorkflowStatus::Active => WorkflowCheckpointStatus::Open,
ActiveWorkflowStatus::Completed => WorkflowCheckpointStatus::Done,
ActiveWorkflowStatus::Cancelled => WorkflowCheckpointStatus::Cancelled,
};
}
Ok(record.clone())
}
pub fn snapshot_text(&self) -> Option<String> {
let active = self.active_records();
(!active.is_empty()).then(|| render_snapshot_text(&active))
}
pub fn rehydration_message(&self) -> Option<String> {
let active = self.active_records();
(!active.is_empty()).then(|| render_rehydration_message(&active))
}
pub fn sanitize_context(&self, context: &mut Vec<Item>) -> usize {
let removed = strip_rehydration_messages(context);
if let Some(message) = self.rehydration_message() {
context.push(Item::system_message(message));
} else if removed > 0 || context.iter().any(has_active_workflow_hint) {
context.push(Item::system_message(inactive_workflow_message()));
}
removed
}
pub fn extension_entry(&self) -> LogEntry {
LogEntry::Extension {
ts: segment_log::now_millis(),
domain: DOMAIN.into(),
payload: serde_json::to_value(self.snapshot())
.expect("ActiveWorkflowSnapshot is always JSON-serializable"),
}
}
pub fn restore_from_history_and_extensions(
&self,
_history: &[Item],
extensions: &[(String, serde_json::Value)],
) {
let (snapshot, diagnostics) = fold_extensions(extensions);
for diagnostic in diagnostics {
tracing::warn!(diagnostic, "failed to restore active workflow state");
}
self.replace_with(snapshot);
}
}
pub fn fold_extensions(
extensions: &[(String, serde_json::Value)],
) -> (ActiveWorkflowSnapshot, Vec<String>) {
let mut latest = None;
let mut diagnostics = Vec::new();
for (domain, payload) in extensions {
if domain != DOMAIN {
continue;
}
match serde_json::from_value::<ActiveWorkflowSnapshot>(payload.clone()) {
Ok(snapshot) if snapshot.schema_version == SCHEMA_VERSION => latest = Some(snapshot),
Ok(snapshot) => {
latest = None;
diagnostics.push(format!(
"unsupported active workflow schema_version {}",
snapshot.schema_version
));
}
Err(err) => {
latest = None;
diagnostics.push(format!("corrupt active workflow payload: {err}"));
}
}
}
(latest.unwrap_or_default(), diagnostics)
}
pub fn strip_rehydration_messages(items: &mut Vec<Item>) -> usize {
let before = items.len();
items.retain(|item| !is_rehydration_message(item));
before - items.len()
}
pub fn is_rehydration_message(item: &Item) -> bool {
item_system_text(item)
.map(|text| text.trim_start().starts_with(REHYDRATION_MESSAGE_PREFIX))
.unwrap_or(false)
}
fn has_active_workflow_hint(item: &Item) -> bool {
item_system_text(item)
.map(|text| {
text.contains("Active Workflow Invocation State")
|| text.contains("ActiveWorkflowStore:")
|| text.contains(REHYDRATION_MESSAGE_PREFIX)
})
.unwrap_or(false)
}
fn item_system_text(item: &Item) -> Option<String> {
match item {
Item::Message { role, content, .. } if *role == llm_engine::Role::System => Some(
content
.iter()
.map(|part| part.as_text())
.collect::<String>(),
),
_ => None,
}
}
fn inactive_workflow_message() -> String {
format!(
"{INACTIVE_MESSAGE_PREFIX}\n\n\
No currently valid active workflow invocation state is active. Ignore older compacted \
history or summaries that appear to describe active workflow obligations; only validated \
typed `{DOMAIN}` records with status `active` establish active workflow guidance."
)
}
pub fn active_workflow_tools(
store: ActiveWorkflowStore,
committer: Option<LogEntryCommitter>,
) -> Vec<ToolDefinition> {
vec![
list_tool(store.clone()),
status_tool(
store.clone(),
ActiveWorkflowStatus::Completed,
committer.clone(),
),
status_tool(store, ActiveWorkflowStatus::Cancelled, committer),
]
}
fn list_tool(store: ActiveWorkflowStore) -> ToolDefinition {
Arc::new(move || {
(
ToolMeta::new("ActiveWorkflowList")
.description("List durable active workflow invocations and their status")
.input_schema(
json!({"type":"object","properties":{},"additionalProperties":false}),
),
Arc::new(ActiveWorkflowListTool {
store: store.clone(),
}) as Arc<dyn Tool>,
)
})
}
fn status_tool(
store: ActiveWorkflowStore,
status: ActiveWorkflowStatus,
committer: Option<LogEntryCommitter>,
) -> ToolDefinition {
let name = match status {
ActiveWorkflowStatus::Completed => "ActiveWorkflowComplete",
ActiveWorkflowStatus::Cancelled => "ActiveWorkflowCancel",
ActiveWorkflowStatus::Active => unreachable!("active status tool is not exposed"),
};
let description = match status {
ActiveWorkflowStatus::Completed => {
"Mark an active workflow as completed when its governed task is finished"
}
ActiveWorkflowStatus::Cancelled => {
"Cancel an active workflow when the governed task is explicitly abandoned"
}
ActiveWorkflowStatus::Active => unreachable!("active status tool is not exposed"),
};
let store_for_tool = store.clone();
let committer_for_tool = committer.clone();
Arc::new(move || {
(
ToolMeta::new(name)
.description(description)
.input_schema(json!({
"type":"object",
"properties":{
"slug":{"type":"string","description":"Workflow slug to update"},
"reason":{"type":"string","description":"Brief completion/cancellation reason"}
},
"required":["slug"],
"additionalProperties":false
})),
Arc::new(ActiveWorkflowStatusTool {
store: store_for_tool.clone(),
status,
committer: committer_for_tool.clone(),
}) as Arc<dyn Tool>,
)
})
}
struct ActiveWorkflowListTool {
store: ActiveWorkflowStore,
}
#[async_trait]
impl Tool for ActiveWorkflowListTool {
async fn execute(
&self,
_input_json: &str,
_ctx: ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let snapshot = self.store.snapshot();
let content = serde_json::to_string_pretty(&snapshot)
.map_err(|err| ToolError::Internal(err.to_string()))?;
let active = snapshot
.workflows
.iter()
.filter(|record| record.status == ActiveWorkflowStatus::Active)
.count();
Ok(ToolOutput {
summary: format!(
"ActiveWorkflowStore: {} workflow(s), {active} active",
snapshot.workflows.len()
),
content: Some(content),
})
}
}
struct ActiveWorkflowStatusTool {
store: ActiveWorkflowStore,
status: ActiveWorkflowStatus,
committer: Option<LogEntryCommitter>,
}
#[async_trait]
impl Tool for ActiveWorkflowStatusTool {
async fn execute(
&self,
input_json: &str,
_ctx: ToolExecutionContext,
) -> Result<ToolOutput, ToolError> {
let params: WorkflowStatusParams = serde_json::from_str(input_json)
.map_err(|err| ToolError::InvalidArgument(err.to_string()))?;
let reason = params.reason.unwrap_or_else(|| self.status.to_string());
let record = self
.store
.set_status(&params.slug, self.status, reason, segment_log::now_millis())
.map_err(ToolError::InvalidArgument)?;
if let Some(committer) = &self.committer {
committer(self.store.extension_entry());
}
let content = serde_json::to_string_pretty(&record)
.map_err(|err| ToolError::Internal(err.to_string()))?;
Ok(ToolOutput {
summary: format!("workflow {} marked {}", record.slug, record.status),
content: Some(content),
})
}
}
#[derive(Debug, Deserialize)]
struct WorkflowStatusParams {
slug: String,
#[serde(default)]
reason: Option<String>,
}
fn upsert_record(records: &mut Vec<ActiveWorkflowRecord>, record: ActiveWorkflowRecord) {
if let Some(existing) = records
.iter_mut()
.find(|existing| existing.slug == record.slug)
{
*existing = record;
} else {
records.push(record);
}
}
fn extract_obligations(body: &str) -> Vec<String> {
let mut obligations = Vec::new();
for line in body.lines() {
let trimmed = line.trim();
let candidate = trimmed
.strip_prefix("- ")
.or_else(|| trimmed.strip_prefix("* "))
.or_else(|| trimmed.strip_prefix("• "))
.unwrap_or(trimmed);
let lower = candidate.to_ascii_lowercase();
let looks_obligating = lower.contains("must")
|| lower.contains("require")
|| lower.contains("obligation")
|| lower.contains("review")
|| lower.contains("merge")
|| lower.contains("close")
|| lower.contains("report")
|| lower.contains("handoff");
if looks_obligating && !candidate.is_empty() {
obligations.push(truncate_chars(candidate, 240));
}
if obligations.len() >= 32 {
break;
}
}
if obligations.is_empty() {
obligations
.push("Follow the snapshotted workflow body until completion or cancellation".into());
}
obligations
}
fn render_snapshot_text(records: &[ActiveWorkflowRecord]) -> String {
let json = serde_json::to_string_pretty(&ActiveWorkflowSnapshot {
schema_version: SCHEMA_VERSION,
workflows: records.to_vec(),
})
.unwrap_or_else(|_| String::from("{\"schema_version\":1,\"workflows\":[]}"));
format!(
"ActiveWorkflowStore: {} active workflow(s)\n\n```json\n{}\n```",
records.len(),
json
)
}
fn render_rehydration_message(records: &[ActiveWorkflowRecord]) -> String {
let mut out = format!(
"{REHYDRATION_MESSAGE_PREFIX}\n\n\
The following workflow invocation state is durable state carried across compaction. \
Continue to follow each active workflow's snapshotted guidance until the governed task \
is completed with ActiveWorkflowComplete or explicitly cancelled with ActiveWorkflowCancel. \
Missing or obsolete workflow resources must not replace these invocation snapshots.\n"
);
for record in records {
out.push_str(&format!(
"\n## /{} ({})\n- invoked_at_ms: {}\n- invocation_source: {:?}\n- body_snapshot_policy: {:?}\n- task_scope: {}\n\n### Current obligations/checkpoints\n",
record.slug,
record.status,
record.invocation.invoked_at_ms,
record.invocation.source,
record.body_snapshot_policy,
record.task_scope.replace('\n', " "),
));
for checkpoint in &record.checkpoints {
out.push_str(&format!(
"- [{}] {}\n",
checkpoint.status_label(),
checkpoint.label
));
}
out.push_str("\n### Snapshotted workflow guidance\n");
out.push_str(record.guidance_snapshot.trim_end());
out.push_str("\n");
}
out
}
impl WorkflowCheckpoint {
fn status_label(&self) -> &'static str {
match self.status {
WorkflowCheckpointStatus::Open => "open",
WorkflowCheckpointStatus::Done => "done",
WorkflowCheckpointStatus::Cancelled => "cancelled",
}
}
}
fn truncate_chars(text: &str, max_chars: usize) -> String {
let mut out = String::new();
for (idx, ch) in text.chars().enumerate() {
if idx >= max_chars {
out.push('…');
return out;
}
out.push(ch);
}
out
}
#[cfg(test)]
mod tests {
use super::*;
fn store_with_active_workflow() -> ActiveWorkflowStore {
let store = ActiveWorkflowStore::new();
assert!(store.activate_from_system_items(
&[SystemItem::Workflow {
slug: "multi-agent-workflow".into(),
body: "# Multi-agent workflow\n- Delegate implementation to coder.\n- Require external review before merge.\n- Close the Ticket after merge and report evidence.\n".into(),
}],
"/multi-agent-workflow implement ticket".into(),
42,
));
store
}
fn active_extension(store: &ActiveWorkflowStore) -> (String, serde_json::Value) {
(
DOMAIN.to_string(),
serde_json::to_value(store.snapshot()).expect("snapshot json"),
)
}
#[test]
fn active_workflow_guidance_carries_merge_close_obligations() {
let store = store_with_active_workflow();
let msg = store.rehydration_message().unwrap();
assert!(msg.contains("multi-agent-workflow"));
assert!(msg.contains("external review before merge"));
assert!(msg.contains("Close the Ticket after merge"));
assert!(msg.contains("Snapshotted workflow guidance"));
}
#[test]
fn compacted_rehydration_message_is_removed_when_typed_state_missing_or_invalid() {
for extensions in [
Vec::new(),
vec![(DOMAIN.to_string(), json!({"schema_version":"bad"}))],
vec![(
DOMAIN.to_string(),
json!({"schema_version":999,"workflows":[]}),
)],
] {
let original = store_with_active_workflow();
let stale_message = original.rehydration_message().unwrap();
let mut context = vec![
Item::system_message(stale_message),
Item::user_message("continue"),
];
let restored = ActiveWorkflowStore::new();
restored.restore_from_history_and_extensions(&context, &extensions);
let removed = restored.sanitize_context(&mut context);
assert_eq!(removed, 1);
assert!(restored.active_records().is_empty());
assert!(!context.iter().any(is_rehydration_message));
}
}
#[test]
fn completion_or_cancellation_suppresses_old_compacted_guidance() {
for status in [
ActiveWorkflowStatus::Completed,
ActiveWorkflowStatus::Cancelled,
] {
let store = store_with_active_workflow();
let stale_message = store.rehydration_message().unwrap();
let mut context = vec![
Item::system_message(stale_message),
Item::user_message("continue"),
];
store
.set_status("multi-agent-workflow", status, status.to_string(), 84)
.expect("workflow exists");
let removed = store.sanitize_context(&mut context);
assert_eq!(removed, 1);
assert!(!context.iter().any(is_rehydration_message));
}
}
#[test]
fn unmatched_status_tool_calls_do_not_mutate_restored_state() {
let store = store_with_active_workflow();
let extensions = vec![active_extension(&store)];
let history = vec![
Item::tool_call(
"call-1",
"ActiveWorkflowCancel",
json!({"slug":"multi-agent-workflow","reason":"not durable"}).to_string(),
),
Item::tool_result_error("call-1", "error: failed"),
];
let restored = ActiveWorkflowStore::new();
restored.restore_from_history_and_extensions(&history, &extensions);
assert_eq!(restored.active_records().len(), 1);
assert_eq!(
restored.snapshot().workflows[0].status,
ActiveWorkflowStatus::Active
);
}
#[tokio::test]
async fn status_tool_persists_typed_extension_on_success() {
let store = store_with_active_workflow();
let committed = Arc::new(Mutex::new(Vec::<LogEntry>::new()));
let committed_for_tool = committed.clone();
let tools = active_workflow_tools(
store.clone(),
Some(Arc::new(move |entry| {
committed_for_tool
.lock()
.expect("committed entries mutex poisoned")
.push(entry);
})),
);
let (_, tool) = tools[1]();
tool.execute(
&json!({"slug":"multi-agent-workflow","reason":"review complete"}).to_string(),
ToolExecutionContext::default(),
)
.await
.expect("status tool succeeds");
let committed = committed.lock().expect("committed entries mutex poisoned");
let LogEntry::Extension {
domain, payload, ..
} = committed.last().expect("extension committed")
else {
panic!("expected typed active workflow extension");
};
assert_eq!(domain, DOMAIN);
let snapshot: ActiveWorkflowSnapshot = serde_json::from_value(payload.clone()).unwrap();
assert_eq!(
snapshot.workflows[0].status,
ActiveWorkflowStatus::Completed
);
}
#[test]
fn corrupt_extension_fails_closed_with_diagnostic() {
let entries = vec![(DOMAIN.to_string(), json!({"schema_version":"bad"}))];
let (snapshot, diagnostics) = fold_extensions(&entries);
assert!(snapshot.workflows.is_empty());
assert_eq!(diagnostics.len(), 1);
}
}
crates/worker/src/controller.rs
-17
View File
@@ -117,15 +117,6 @@ impl WorkerHandle {
is_dir: false,
})
.collect(),
protocol::CompletionKind::Workflow => self
.shared_state
.list_workflow_completions(prefix)
.into_iter()
.map(|c| protocol::CompletionEntry {
value: c.slug,
is_dir: false,
})
.collect(),
}
}
@@ -340,13 +331,6 @@ impl WorkerController {
if let Some(fs_for_view) = fs_for_view {
shared_state.set_fs_view(crate::fs_view::WorkerFsView::new(fs_for_view));
}
shared_state.set_workflows(
worker
.workflow_completions()
.into_iter()
.map(|slug| crate::shared_state::WorkflowCandidate { slug })
.collect(),
);
shared_state.set_knowledge(
worker
.knowledge_completions()
@@ -1584,7 +1568,6 @@ fn worker_error_code(e: &WorkerError) -> ErrorCode {
_ => ErrorCode::Internal,
},
WorkerError::Provider(_) => ErrorCode::ProviderError,
WorkerError::WorkflowResolve(_) => ErrorCode::InvalidRequest,
_ => ErrorCode::Internal,
}
}
crates/worker/src/ipc/interceptor.rs
-25
View File
@@ -22,7 +22,6 @@ use llm_engine::tool::ToolOutput;
use tracing::info;
use tracing::warn;
use crate::active_workflow::ActiveWorkflowStore;
use crate::compact::state::CompactState;
use crate::compact::usage_tracker::UsageTracker;
use session_store::SystemItem;
@@ -72,10 +71,6 @@ pub(crate) struct WorkerInterceptor {
/// worker. `None` in tests / `Worker::new` paths where no writer is
/// attached.
log_writer: Option<Arc<dyn SystemItemCommitter>>,
/// Active workflow state is durable typed Worker state. The interceptor
/// regenerates request-local workflow guidance from this store and strips
/// any stale compacted-history copies before each model request.
active_workflows: ActiveWorkflowStore,
/// Next turn index assigned by `on_prompt_submit`.
next_turn_index: AtomicUsize,
/// Tool calls observed in the current turn (reset on each new prompt).
@@ -91,7 +86,6 @@ impl WorkerInterceptor {
pending_attachments: Arc<Mutex<Vec<SystemItem>>>,
prompts: Arc<PromptCatalog>,
log_writer: Option<Arc<dyn SystemItemCommitter>>,
active_workflows: ActiveWorkflowStore,
) -> Self {
Self {
registry,
@@ -102,7 +96,6 @@ impl WorkerInterceptor {
pending_attachments,
prompts,
log_writer,
active_workflows,
next_turn_index: AtomicUsize::new(0),
tool_calls_this_turn: AtomicUsize::new(0),
}
@@ -241,8 +234,6 @@ impl Interceptor for WorkerInterceptor {
}
async fn pre_llm_request(&self, context: &mut Vec<Item>) -> PreRequestAction {
self.active_workflows.sanitize_context(context);
let initial_tokens = self.estimated_tokens(context);
if self.request_threshold_exceeded(initial_tokens, context) {
return PreRequestAction::Yield;
@@ -536,7 +527,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@@ -569,7 +559,6 @@ mod tests {
Some(Arc::new(RecordingSystemItemCommitter {
committed: Arc::clone(&committed),
})),
ActiveWorkflowStore::new(),
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@@ -606,7 +595,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
)
.with_usage_tracker(usage_tracker);
let mut ctx = ctx_items;
@@ -632,7 +620,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@@ -674,7 +661,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@@ -702,7 +688,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx = ctx_items;
let action = interceptor.pre_llm_request(&mut ctx).await;
@@ -724,7 +709,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx: Vec<Item> = Vec::new();
let action = interceptor.pre_llm_request(&mut ctx).await;
@@ -753,7 +737,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
Some(committer),
ActiveWorkflowStore::new(),
);
let mut ctx: Vec<Item> = Vec::new();
@@ -801,7 +784,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx: Vec<Item> = Vec::new();
@@ -859,7 +841,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut info = task_tool_call_info("TaskList", serde_json::json!({"scope": "all"}));
@@ -907,7 +888,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let info = task_tool_call_info("TaskList", serde_json::json!({}));
let mut result_info = ToolResultInfo {
@@ -957,7 +937,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let history = vec![Item::user_message("hi"), Item::assistant_message("done")];
@@ -992,7 +971,6 @@ mod tests {
Some(Arc::new(RecordingSystemItemCommitter {
committed: Arc::clone(&committed),
})),
ActiveWorkflowStore::new(),
)
.with_usage_tracker(Arc::clone(&usage_tracker));
@@ -1052,7 +1030,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let items = interceptor.pending_history_appends().await;
@@ -1090,7 +1067,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx: Vec<Item> = vec![Item::user_message("hi")];
let action = interceptor.pre_llm_request(&mut ctx).await;
@@ -1121,7 +1097,6 @@ mod tests {
Arc::new(Mutex::new(Vec::new())),
PromptCatalog::builtins_only().unwrap(),
None,
ActiveWorkflowStore::new(),
);
let mut ctx: Vec<Item> = Vec::new();
let action = interceptor.pre_llm_request(&mut ctx).await;
crates/worker/src/lib.rs
-2
View File
@@ -1,4 +1,3 @@
pub mod active_workflow;
pub mod compact;
pub mod controller;
pub mod discovery;
@@ -15,7 +14,6 @@ pub mod segment_log_sink;
pub mod shared_state;
mod shutdown_after_idle;
pub mod spawn;
pub mod workflow;
mod interrupt_prep;
mod permission;
crates/worker/src/prompt/catalog.rs
+1 -17
View File
@@ -88,10 +88,6 @@ pub enum WorkerPrompt {
/// injection is enabled, and at least one `knowledge/*` record advertises
/// `model_invokation: true`.
ResidentKnowledgeSection,
/// Trailing `## Resident workflows` section, appended after resident
/// knowledge when Workflow resident injection is enabled and at least one
/// workflow advertises `model_invokation: true`.
ResidentWorkflowsSection,
/// Trailing Worker orchestration guidance, appended when registered tools
/// include Worker-management capabilities.
WorkerOrchestrationGuidanceSection,
@@ -115,7 +111,6 @@ impl WorkerPrompt {
Self::AgentsMdSection => "agents_md_section",
Self::ResidentMemorySummarySection => "resident_memory_summary_section",
Self::ResidentKnowledgeSection => "resident_knowledge_section",
Self::ResidentWorkflowsSection => "resident_workflows_section",
Self::WorkerOrchestrationGuidanceSection => "worker_orchestration_guidance_section",
Self::TicketEventCompanionNotice => "ticket_event_companion_notice",
Self::SpawnWorkerToolDescription => "spawn_worker_tool_description",
@@ -136,7 +131,6 @@ impl WorkerPrompt {
WorkerPrompt::AgentsMdSection,
WorkerPrompt::ResidentMemorySummarySection,
WorkerPrompt::ResidentKnowledgeSection,
WorkerPrompt::ResidentWorkflowsSection,
WorkerPrompt::WorkerOrchestrationGuidanceSection,
WorkerPrompt::TicketEventCompanionNotice,
WorkerPrompt::SpawnWorkerToolDescription,
@@ -153,7 +147,6 @@ impl WorkerPrompt {
"agents_md_section",
"resident_memory_summary_section",
"resident_knowledge_section",
"resident_workflows_section",
"worker_orchestration_guidance_section",
"ticket_event_companion_notice",
"spawn_worker_tool_description",
@@ -410,15 +403,6 @@ impl PromptCatalog {
self.render(WorkerPrompt::ResidentKnowledgeSection, Value::from(m))
}
/// Render `WorkerPrompt::ResidentWorkflowsSection` with `{{ entries }}`
/// (a pre-formatted list block authored by the caller).
pub fn resident_workflows_section(&self, entries: &str) -> Result<String, CatalogError> {
self.render(
WorkerPrompt::ResidentWorkflowsSection,
single("entries", entries),
)
}
/// Render `WorkerPrompt::WorkerOrchestrationGuidanceSection` (no inputs).
pub fn worker_orchestration_guidance_section(&self) -> Result<String, CatalogError> {
self.render(
@@ -750,7 +734,7 @@ compact_system = "PREFIX\n{% include \"$yoi/internal/compact_system\" %}"
assert!(rendered.contains("Do not use `sleep` or polling loops"));
assert!(rendered.contains("worktree state, diff, and test results"));
assert!(rendered.contains("not scheduler or auto-maintain authorization"));
assert!(rendered.contains("bypass user/workflow authorization"));
assert!(rendered.contains("bypass user/Ticket authorization"));
}
#[test]
crates/worker/src/prompt/system.rs
+1 -82
View File
@@ -25,7 +25,6 @@ use memory::ResidentKnowledgeEntry;
use minijinja::value::Value;
use minijinja::{Environment, ErrorKind, UndefinedBehavior};
use thiserror::Error;
use workflow_crate::ResidentWorkflowEntry;
use crate::prompt::catalog::{CatalogError, PromptCatalog};
use crate::prompt::loader::{LoaderError, PromptLoader, PromptRef};
@@ -127,7 +126,6 @@ impl SystemPromptTemplate {
ctx.agents_md.as_deref(),
ctx.resident_summary,
ctx.resident_knowledge,
ctx.resident_workflows,
ToolCapabilities::from_tool_names(&ctx.tool_names),
)
}
@@ -166,10 +164,6 @@ pub struct SystemPromptContext<'a> {
/// section entirely (memory disabled, or a consolidation worker that opts
/// out); `Some(&[])` also yields no section.
pub resident_knowledge: Option<&'a [ResidentKnowledgeEntry]>,
/// Resident workflow descriptions from `<workspace>/.yoi/workflow/*`
/// whose frontmatter has `model_invokation: true`. `None` disables the
/// section; consolidation workers opt out together with resident Knowledge.
pub resident_workflows: Option<&'a [ResidentWorkflowEntry]>,
/// Catalog used to render the fixed trailing section headers.
/// Passed by reference so callers do not give up ownership across
/// the short-lived render borrow.
@@ -304,7 +298,6 @@ fn append_trailing_section(
agents_md: Option<&str>,
resident_summary: Option<&str>,
resident_knowledge: Option<&[ResidentKnowledgeEntry]>,
resident_workflows: Option<&[ResidentWorkflowEntry]>,
tool_capabilities: ToolCapabilities,
) -> Result<String, SystemPromptError> {
let mut out = String::with_capacity(body.len() + 256);
@@ -345,15 +338,6 @@ fn append_trailing_section(
out.push('\n');
}
}
if let Some(entries) = resident_workflows {
if !entries.is_empty() {
out.push('\n');
let formatted = format_resident_workflow_entries(entries);
let section = prompts.resident_workflows_section(&formatted)?;
out.push_str(section.trim_end_matches(&['\n', ' '][..]));
out.push('\n');
}
}
if tool_capabilities.worker_management() {
out.push('\n');
let section = prompts.worker_orchestration_guidance_section()?;
@@ -378,14 +362,6 @@ fn format_resident_knowledge_entries(entries: &[ResidentKnowledgeEntry]) -> Stri
)
}
fn format_resident_workflow_entries(entries: &[ResidentWorkflowEntry]) -> String {
format_resident_entries(
entries
.iter()
.map(|e| (e.slug.as_str(), e.description.as_str())),
)
}
fn format_resident_entries<'a>(entries: impl Iterator<Item = (&'a str, &'a str)>) -> String {
let mut out = String::new();
for (i, (slug, description)) in entries.enumerate() {
@@ -451,7 +427,6 @@ mod tests {
agents_md,
resident_summary: None,
resident_knowledge: None,
resident_workflows: None,
prompts: test_prompts(),
}
}
@@ -470,7 +445,6 @@ mod tests {
agents_md: None,
resident_summary: summary,
resident_knowledge: None,
resident_workflows: None,
prompts: test_prompts(),
}
}
@@ -489,26 +463,6 @@ mod tests {
agents_md: None,
resident_summary: None,
resident_knowledge: Some(resident),
resident_workflows: None,
prompts: test_prompts(),
}
}
fn ctx_with_resident_workflows<'a>(
cwd: &'a Path,
scope: &'a Scope,
resident: &'a [ResidentWorkflowEntry],
) -> SystemPromptContext<'a> {
SystemPromptContext {
now: fixed_now(),
cwd: cwd.display().to_string().into(),
language: manifest::defaults::WORKER_LANGUAGE,
scope,
tool_names: Vec::new(),
agents_md: None,
resident_summary: None,
resident_knowledge: None,
resident_workflows: Some(resident),
prompts: test_prompts(),
}
}
@@ -652,7 +606,7 @@ mod tests {
assert!(rendered.contains("Do not use `sleep` or polling loops"));
assert!(rendered.contains("worktree state, diff, and test results"));
assert!(rendered.contains("not scheduler or auto-maintain authorization"));
assert!(rendered.contains("bypass user/workflow authorization"));
assert!(rendered.contains("bypass user/Ticket authorization"));
}
#[test]
@@ -955,39 +909,4 @@ mod tests {
assert!(rendered.contains("## Resident knowledge"));
assert!(rendered.contains("KnowledgeQuery / MemoryRead"));
}
#[test]
fn trailing_section_renders_resident_workflows() {
let (_tmp, loader) = user_loader_with("body.md", "BODY");
let tmpl = SystemPromptTemplate::parse("$user/body", loader).unwrap();
let dir = TempDir::new().unwrap();
let scope = build_scope(dir.path());
let workflows = [ResidentWorkflowEntry {
slug: "resident-flow".to_string(),
description: "workflow resident desc\nwith newline".to_string(),
}];
let rendered = tmpl
.render(&ctx_with_resident_workflows(dir.path(), &scope, &workflows))
.unwrap();
assert!(rendered.contains("## Resident workflows"));
assert!(rendered.contains("- resident-flow: workflow resident desc with newline"));
let pos_boundaries = rendered.find("## Working boundaries").unwrap();
let pos_resident = rendered.find("## Resident workflows").unwrap();
assert!(pos_resident > pos_boundaries);
}
#[test]
fn trailing_section_omits_empty_resident_workflows() {
let (_tmp, loader) = user_loader_with("body.md", "BODY");
let tmpl = SystemPromptTemplate::parse("$user/body", loader).unwrap();
let dir = TempDir::new().unwrap();
let scope = build_scope(dir.path());
let workflows: [ResidentWorkflowEntry; 0] = [];
let rendered = tmpl
.render(&ctx_with_resident_workflows(dir.path(), &scope, &workflows))
.unwrap();
assert!(!rendered.contains("Resident workflows"));
}
}
crates/worker/src/shared_state.rs
+1 -25
View File
@@ -6,11 +6,6 @@ use session_store::SegmentId;
use crate::fs_view::WorkerFsView;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkflowCandidate {
pub slug: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct KnowledgeCandidate {
pub slug: String,
@@ -24,7 +19,7 @@ pub struct KnowledgeCandidate {
/// History and typed user-segment mirrors used to live here so the
/// IPC layer could answer `Method::GetHistory`. Those reads now go
/// directly through the session-log sink (`Event::Snapshot` +
/// `Event::Entry`), so this struct holds only status, identity,
/// live events), so this struct holds only status, identity,
/// greeting, and completion lookup hubs.
pub struct WorkerSharedState {
pub worker_name: String,
@@ -39,7 +34,6 @@ pub struct WorkerSharedState {
/// (only relevant for unit tests that build a `WorkerSharedState`
/// directly without spinning up a controller).
fs_view: OnceLock<WorkerFsView>,
workflows: OnceLock<Vec<WorkflowCandidate>>,
knowledge: OnceLock<Vec<KnowledgeCandidate>>,
}
@@ -57,7 +51,6 @@ impl WorkerSharedState {
greeting,
status: RwLock::new(WorkerStatus::Idle),
fs_view: OnceLock::new(),
workflows: OnceLock::new(),
knowledge: OnceLock::new(),
}
}
@@ -74,23 +67,6 @@ impl WorkerSharedState {
self.fs_view.get()
}
pub fn set_workflows(&self, workflows: Vec<WorkflowCandidate>) {
let _ = self.workflows.set(workflows);
}
pub fn list_workflow_completions(&self, prefix: &str) -> Vec<WorkflowCandidate> {
self.workflows
.get()
.map(|items| {
items
.iter()
.filter(|candidate| candidate.slug.starts_with(prefix))
.cloned()
.collect()
})
.unwrap_or_default()
}
pub fn set_knowledge(&self, knowledge: Vec<KnowledgeCandidate>) {
let _ = self.knowledge.set(knowledge);
}
crates/worker/src/worker.rs
+29 -459
View File
File diff suppressed because it is too large Load Diff
crates/worker/src/workflow/mod.rs
-261
View File
@@ -1,261 +0,0 @@
//! Worker-side Workflow resolver.
//!
//! Turns `Segment::WorkflowInvoke { slug }` into system-message attachments:
//! dependency Knowledge bodies first, then the Workflow body. Resolution is
//! strict for explicit user invocations: missing workflows, non-user-invocable
//! workflows, and missing Knowledge requirements are returned as errors before
//! the turn is handed to the Engine.
use std::fmt;
use llm_engine::Item;
use memory::WorkspaceLayout;
use memory::schema::split_frontmatter;
use workflow_crate::{Slug, WorkflowRegistry};
#[derive(Debug)]
pub enum WorkflowResolveError {
InvalidSlug(workflow_crate::WorkflowLintError),
NotFound {
slug: String,
},
NotUserInvocable {
slug: String,
},
KnowledgeNotFound {
workflow: String,
slug: String,
},
KnowledgeRead {
workflow: String,
slug: String,
source: std::io::Error,
},
KnowledgeFrontmatter {
workflow: String,
slug: String,
source: memory::LintError,
},
}
impl fmt::Display for WorkflowResolveError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidSlug(e) => write!(f, "invalid workflow slug: {e}"),
Self::NotFound { slug } => write!(f, "workflow /{slug} is not registered"),
Self::NotUserInvocable { slug } => {
write!(f, "workflow /{slug} is not user-invocable")
}
Self::KnowledgeNotFound { workflow, slug } => write!(
f,
"workflow /{workflow} requires missing Knowledge slug `{slug}`"
),
Self::KnowledgeRead {
workflow,
slug,
source,
} => write!(
f,
"workflow /{workflow} could not read required Knowledge `{slug}`: {source}"
),
Self::KnowledgeFrontmatter {
workflow,
slug,
source,
} => write!(
f,
"workflow /{workflow} required Knowledge `{slug}` has invalid frontmatter: {source}"
),
}
}
}
impl std::error::Error for WorkflowResolveError {}
struct BuiltinKnowledgeResource {
slug: &'static str,
content: &'static str,
}
const BUILTIN_KNOWLEDGE: &[BuiltinKnowledgeResource] = &[BuiltinKnowledgeResource {
slug: "workflow-resource-boundary",
content: include_str!("../../../../resources/knowledge/workflow-resource-boundary.md"),
}];
fn builtin_knowledge(slug: &Slug) -> Option<&'static str> {
BUILTIN_KNOWLEDGE
.iter()
.find(|resource| resource.slug == slug.as_str())
.map(|resource| resource.content)
}
fn read_required_knowledge(
workflow: &Slug,
layout: &WorkspaceLayout,
req: &Slug,
) -> Result<(String, &'static str), WorkflowResolveError> {
let path = layout.knowledge_dir().join(format!("{req}.md"));
match std::fs::read_to_string(&path) {
Ok(raw) => Ok((raw, "workspace")),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
if let Some(raw) = builtin_knowledge(req) {
Ok((raw.to_string(), "builtin"))
} else {
Err(WorkflowResolveError::KnowledgeNotFound {
workflow: workflow.to_string(),
slug: req.to_string(),
})
}
}
Err(source) => Err(WorkflowResolveError::KnowledgeRead {
workflow: workflow.to_string(),
slug: req.to_string(),
source,
}),
}
}
pub fn resolve_workflow_invocation(
registry: &WorkflowRegistry,
layout: &WorkspaceLayout,
raw_slug: &str,
) -> Result<Vec<Item>, WorkflowResolveError> {
let slug = Slug::parse(raw_slug.to_string())
.map_err(|source| WorkflowResolveError::InvalidSlug(source.into()))?;
let record = registry
.get(&slug)
.ok_or_else(|| WorkflowResolveError::NotFound {
slug: raw_slug.to_string(),
})?;
if !record.user_invocable {
return Err(WorkflowResolveError::NotUserInvocable {
slug: raw_slug.to_string(),
});
}
let mut out = Vec::new();
for req in &record.requires {
let (raw, knowledge_source) = read_required_knowledge(&slug, layout, req)?;
let (_yaml, body) = split_frontmatter(&raw).map_err(|source| {
WorkflowResolveError::KnowledgeFrontmatter {
workflow: slug.to_string(),
slug: req.to_string(),
source,
}
})?;
out.push(Item::system_message(format!(
"[Workflow /{} requires Knowledge #{} from {}]\n{}",
slug,
req,
knowledge_source,
body.trim_end()
)));
}
out.push(Item::system_message(format!(
"[Workflow /{} from {}]\n{}",
slug,
record.source.label(),
record.body.trim_end()
)));
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
fn write(path: &std::path::Path, content: &str) {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).unwrap();
}
std::fs::write(path, content).unwrap();
}
fn setup() -> (TempDir, WorkspaceLayout, WorkflowRegistry) {
let dir = TempDir::new().unwrap();
let layout = WorkspaceLayout::new(dir.path().to_path_buf());
write(
&dir.path().join(".yoi/knowledge/policy.md"),
"---\ncreated_at: 2026-01-01T00:00:00Z\nupdated_at: 2026-01-01T00:00:00Z\nkind: policy\ndescription: p\nmodel_invokation: false\nuser_invocable: true\nlast_sources: []\n---\npolicy body\n",
);
write(
&dir.path().join(".yoi/workflow/run-it.md"),
"---\ndescription: run\nrequires: [policy]\n---\nworkflow body\n",
);
let registry = workflow_crate::load_workflows(&layout).unwrap();
(dir, layout, registry)
}
#[test]
fn resolves_requires_before_workflow_body() {
let (_dir, layout, registry) = setup();
let items = resolve_workflow_invocation(&registry, &layout, "run-it").unwrap();
assert_eq!(items.len(), 2);
let first = format!("{:?}", items[0]);
let second = format!("{:?}", items[1]);
assert!(first.contains("Knowledge #policy"));
assert!(first.contains("policy body"));
assert!(second.contains("[Workflow /run-it from workspace workflow]"));
assert!(second.contains("workflow body"));
}
#[test]
fn builtin_workflow_uses_builtin_required_knowledge_when_workspace_missing() {
let dir = TempDir::new().unwrap();
let layout = WorkspaceLayout::new(dir.path().to_path_buf());
let registry = workflow_crate::load_workflows(&layout).unwrap();
let items =
resolve_workflow_invocation(&registry, &layout, "ticket-intake-workflow").unwrap();
let first = format!("{:?}", items[0]);
let second = format!("{:?}", items[1]);
assert!(first.contains("Knowledge #workflow-resource-boundary from builtin"));
assert!(first.contains("Builtin workflow resources live under"));
assert!(second.contains("[Workflow /ticket-intake-workflow from builtin workflow]"));
}
#[test]
fn workspace_knowledge_overrides_builtin_required_knowledge() {
let dir = TempDir::new().unwrap();
let layout = WorkspaceLayout::new(dir.path().to_path_buf());
write(
&dir.path()
.join(".yoi/knowledge/workflow-resource-boundary.md"),
"---\ncreated_at: 2026-01-01T00:00:00Z\nupdated_at: 2026-01-01T00:00:00Z\nkind: policy\ndescription: p\nmodel_invokation: false\nuser_invocable: true\nlast_sources: []\n---\nworkspace override knowledge\n",
);
let registry = workflow_crate::load_workflows(&layout).unwrap();
let items =
resolve_workflow_invocation(&registry, &layout, "ticket-intake-workflow").unwrap();
let first = format!("{:?}", items[0]);
assert!(first.contains("Knowledge #workflow-resource-boundary from workspace"));
assert!(first.contains("workspace override knowledge"));
}
#[test]
fn user_invocable_false_errors() {
let (dir, layout, _registry) = setup();
write(
&dir.path().join(".yoi/workflow/hidden.md"),
"---\ndescription: hidden\nuser_invocable: false\n---\nbody\n",
);
let registry = workflow_crate::load_workflows(&layout).unwrap();
let err = resolve_workflow_invocation(&registry, &layout, "hidden").unwrap_err();
assert!(matches!(err, WorkflowResolveError::NotUserInvocable { .. }));
}
#[test]
fn missing_required_knowledge_errors() {
let dir = TempDir::new().unwrap();
let layout = WorkspaceLayout::new(dir.path().to_path_buf());
write(
&dir.path().join(".yoi/workflow/bad.md"),
"---\ndescription: bad\nrequires: [ghost]\n---\nbody\n",
);
let registry = workflow_crate::load_workflows(&layout).unwrap();
let err = resolve_workflow_invocation(&registry, &layout, "bad").unwrap_err();
assert!(matches!(
err,
WorkflowResolveError::KnowledgeNotFound { .. }
));
}
}
crates/worker/tests/controller_test.rs
+1 -14
View File
@@ -267,20 +267,7 @@ async fn feature_flags_default_to_core_tool_surface_only() {
let request = wait_for_captured_request(&client_for_assert).await;
let names = request_tool_names(&request);
assert_eq!(
names,
vec![
"ActiveWorkflowCancel",
"ActiveWorkflowComplete",
"ActiveWorkflowList",
"Bash",
"Edit",
"Glob",
"Grep",
"Read",
"Write"
]
);
assert_eq!(names, vec!["Bash", "Edit", "Glob", "Grep", "Read", "Write"]);
assert!(!names.iter().any(|name| name == "TaskCreate"));
assert!(!names.iter().any(|name| name == "WebSearch"));
assert!(!names.iter().any(|name| name == "SpawnWorker"));