everruns_core/capabilities/
delegation_result.rs

1use crate::session_task::{
2    NewTaskMessage, SessionTask, SessionTaskUpdate, TaskMessageDirection, TaskMessagePart,
3    task_result_path,
4};
5use crate::tools::{Tool, ToolExecutionResult};
6use crate::traits::{SessionFileSystem, SessionStore, ToolContext};
7use crate::typed_id::{SessionId, WorkspaceId};
8use async_trait::async_trait;
9use serde_json::{Value, json};
10use std::sync::Arc;
11
12pub(crate) const RESULT_SCHEMA_SPEC_KEY: &str = "result_schema";
13pub(crate) const MESSAGE_SCHEMA_SPEC_KEY: &str = "message_schema";
14
15pub(crate) fn declared_result_schema(task: &SessionTask) -> Option<&Value> {
16    task.spec
17        .get(RESULT_SCHEMA_SPEC_KEY)
18        .filter(|schema| schema.is_object())
19}
20
21pub(crate) fn declared_message_schema(task: &SessionTask) -> Option<&Value> {
22    task.spec
23        .get(MESSAGE_SCHEMA_SPEC_KEY)
24        .filter(|schema| schema.is_object())
25}
26
27fn normalize_optional_schema(
28    arguments: &Value,
29    key: &str,
30) -> Result<Option<Value>, ToolExecutionResult> {
31    let Some(schema) = arguments.get(key).filter(|value| !value.is_null()) else {
32        return Ok(None);
33    };
34    if !schema.is_object() {
35        return Err(ToolExecutionResult::tool_error(format!(
36            "{key} must be a JSON Schema object when provided."
37        )));
38    }
39    Ok(Some(schema.clone()))
40}
41
42pub(crate) fn normalize_result_schema(
43    arguments: &Value,
44) -> Result<Option<Value>, ToolExecutionResult> {
45    normalize_optional_schema(arguments, RESULT_SCHEMA_SPEC_KEY)
46}
47
48pub(crate) fn normalize_message_schema(
49    arguments: &Value,
50) -> Result<Option<Value>, ToolExecutionResult> {
51    normalize_optional_schema(arguments, MESSAGE_SCHEMA_SPEC_KEY)
52}
53
54pub(crate) fn schema_validation_errors(schema: &Value, value: &Value) -> Vec<String> {
55    let validator = match jsonschema::draft202012::options()
56        .should_validate_formats(true)
57        .build(schema)
58    {
59        Ok(validator) => validator,
60        Err(error) => return vec![format!("result schema is invalid: {error}")],
61    };
62
63    validator
64        .iter_errors(value)
65        .map(|error| {
66            let path = error.instance_path().to_string();
67            let path = if path.is_empty() {
68                "$".to_string()
69            } else {
70                path
71            };
72            format!("{path} {error}")
73        })
74        .collect()
75}
76
77const MAX_TASK_SUMMARY_CHARS: usize = 2_048;
78
79pub(crate) fn truncate_summary(text: &str) -> String {
80    let mut chars = text.chars();
81    let truncated: String = chars.by_ref().take(MAX_TASK_SUMMARY_CHARS).collect();
82    if chars.next().is_some() {
83        format!("{truncated}\n[truncated]")
84    } else {
85        truncated
86    }
87}
88
89pub(crate) async fn task_for_child_session(
90    child_session_id: SessionId,
91    session_store: &dyn SessionStore,
92    task_registry: &dyn crate::session_task::SessionTaskRegistry,
93) -> crate::error::Result<Option<(SessionTask, WorkspaceId)>> {
94    let Some(child) = session_store.get_session(child_session_id).await? else {
95        return Ok(None);
96    };
97    let Some(parent_session_id) = child.parent_session_id.or(child.forked_from_session_id) else {
98        return Ok(None);
99    };
100    let Some(parent) = session_store.get_session(parent_session_id).await? else {
101        return Ok(None);
102    };
103    let task = task_registry
104        .list(parent_session_id, None)
105        .await?
106        .into_iter()
107        .find(|task| task.links.child_session_id == Some(child_session_id));
108    Ok(task.map(|task| (task, parent.workspace_id)))
109}
110
111pub struct ReportResultTool {
112    parent_session_id: SessionId,
113    parent_workspace_id: WorkspaceId,
114    child_session_id: SessionId,
115    task_id: String,
116    result_schema: Value,
117    file_store: Option<Arc<dyn SessionFileSystem>>,
118}
119
120impl ReportResultTool {
121    pub fn new(
122        parent_session_id: SessionId,
123        parent_workspace_id: WorkspaceId,
124        child_session_id: SessionId,
125        task_id: String,
126        result_schema: Value,
127    ) -> Self {
128        Self {
129            parent_session_id,
130            parent_workspace_id,
131            child_session_id,
132            task_id,
133            result_schema,
134            file_store: None,
135        }
136    }
137
138    pub fn with_file_store(mut self, file_store: Arc<dyn SessionFileSystem>) -> Self {
139        self.file_store = Some(file_store);
140        self
141    }
142}
143
144#[async_trait]
145impl Tool for ReportResultTool {
146    fn narrate(
147        &self,
148        tool_call: &crate::tool_types::ToolCall,
149        phase: crate::tool_narration::ToolNarrationPhase,
150        locale: Option<&str>,
151        _ctx: crate::tool_narration::ToolNarrationContext<'_>,
152    ) -> Option<String> {
153        crate::tool_narration::narrate_delegation_result(&tool_call.name, phase, locale)
154    }
155
156    fn name(&self) -> &str {
157        "report_result"
158    }
159
160    fn display_name(&self) -> Option<&str> {
161        Some("Report Result")
162    }
163
164    fn description(&self) -> &str {
165        "Submit the final structured result for this delegated task. The call arguments must match the declared result schema."
166    }
167
168    fn parameters_schema(&self) -> Value {
169        self.result_schema.clone()
170    }
171
172    async fn execute(&self, _arguments: Value) -> ToolExecutionResult {
173        ToolExecutionResult::tool_error(
174            "report_result requires context. This tool must be executed with session context.",
175        )
176    }
177
178    async fn execute_with_context(
179        &self,
180        arguments: Value,
181        context: &ToolContext,
182    ) -> ToolExecutionResult {
183        let errors = schema_validation_errors(&self.result_schema, &arguments);
184        if !errors.is_empty() {
185            return ToolExecutionResult::tool_error(format!(
186                "report_result arguments do not match result_schema: {}",
187                errors.join("; ")
188            ));
189        }
190        let Some(registry) = context.session_task_registry.as_ref() else {
191            return ToolExecutionResult::tool_error(
192                "report_result requires session_task_registry context",
193            );
194        };
195        let Some(file_store) = self.file_store.as_ref().or(context.file_store.as_ref()) else {
196            return ToolExecutionResult::tool_error("report_result requires file_store context");
197        };
198
199        if context.session_id != self.child_session_id {
200            return ToolExecutionResult::tool_error(
201                "report_result can only be called from the linked child session",
202            );
203        }
204        let task = match registry.get(self.parent_session_id, &self.task_id).await {
205            Ok(Some(task)) => task,
206            Ok(None) => return ToolExecutionResult::tool_error("report_result task was not found"),
207            Err(error) => return ToolExecutionResult::internal_error(error),
208        };
209        if task.links.child_session_id != Some(self.child_session_id) {
210            return ToolExecutionResult::tool_error(
211                "report_result child session is no longer linked to this task",
212            );
213        }
214        if task.state.is_terminal() {
215            return ToolExecutionResult::tool_error(
216                "report_result is closed because the subagent task is terminal",
217            );
218        }
219        if task.result_path.is_some() {
220            return ToolExecutionResult::tool_error(
221                "report_result was already recorded for this subagent task",
222            );
223        }
224
225        let path = task_result_path(&self.task_id);
226        let content = match serde_json::to_string_pretty(&arguments) {
227            Ok(content) => content,
228            Err(error) => return ToolExecutionResult::internal_error(error),
229        };
230        let parent_workspace_key = SessionId::from_uuid(self.parent_workspace_id.uuid());
231        if let Err(error) = file_store
232            .write_file(parent_workspace_key, &path, &content, "utf-8")
233            .await
234        {
235            return ToolExecutionResult::internal_error(error);
236        }
237        if let Err(error) = registry
238            .update(
239                self.parent_session_id,
240                &self.task_id,
241                SessionTaskUpdate {
242                    // Fence result writes to the active attempt and repeat the
243                    // observed non-terminal state so a task that finalized after
244                    // our read rejects this stale content mutation too.
245                    expected_attempt: Some(task.attempt),
246                    state: Some(task.state),
247                    result_path: Some(path.clone()),
248                    summary: Some(truncate_summary(&content)),
249                    ..Default::default()
250                },
251            )
252            .await
253        {
254            return ToolExecutionResult::internal_error(error);
255        }
256        ToolExecutionResult::success(json!({
257            "status": "recorded",
258            "task_id": self.task_id,
259            "result_path": path,
260        }))
261    }
262
263    fn requires_context(&self) -> bool {
264        true
265    }
266}
267
268pub struct ReportTaskProgressTool {
269    parent_session_id: SessionId,
270    task_id: String,
271    task_attempt: i32,
272    message_schema: Value,
273}
274
275impl ReportTaskProgressTool {
276    pub fn new(
277        parent_session_id: SessionId,
278        task_id: String,
279        task_attempt: i32,
280        message_schema: Value,
281    ) -> Self {
282        Self {
283            parent_session_id,
284            task_id,
285            task_attempt,
286            message_schema,
287        }
288    }
289}
290
291#[async_trait]
292impl Tool for ReportTaskProgressTool {
293    fn narrate(
294        &self,
295        tool_call: &crate::tool_types::ToolCall,
296        phase: crate::tool_narration::ToolNarrationPhase,
297        locale: Option<&str>,
298        _ctx: crate::tool_narration::ToolNarrationContext<'_>,
299    ) -> Option<String> {
300        crate::tool_narration::narrate_delegation_result(&tool_call.name, phase, locale)
301    }
302
303    fn name(&self) -> &str {
304        "report_task_progress"
305    }
306
307    fn display_name(&self) -> Option<&str> {
308        Some("Report Task Progress")
309    }
310
311    fn description(&self) -> &str {
312        "Post a structured progress message for this delegated task. The call arguments must match the declared message schema."
313    }
314
315    fn parameters_schema(&self) -> Value {
316        self.message_schema.clone()
317    }
318
319    async fn execute(&self, _arguments: Value) -> ToolExecutionResult {
320        ToolExecutionResult::tool_error(
321            "report_task_progress requires context. This tool must be executed with session context.",
322        )
323    }
324
325    async fn execute_with_context(
326        &self,
327        arguments: Value,
328        context: &ToolContext,
329    ) -> ToolExecutionResult {
330        let errors = schema_validation_errors(&self.message_schema, &arguments);
331        if !errors.is_empty() {
332            return ToolExecutionResult::tool_error(format!(
333                "report_task_progress arguments do not match message_schema: {}",
334                errors.join("; ")
335            ));
336        }
337        let Some(registry) = context.session_task_registry.as_ref() else {
338            return ToolExecutionResult::tool_error(
339                "report_task_progress requires session_task_registry context",
340            );
341        };
342        let stored = match registry
343            .record_message(
344                self.parent_session_id,
345                &self.task_id,
346                NewTaskMessage {
347                    direction: TaskMessageDirection::Outbound,
348                    content: vec![TaskMessagePart::Data {
349                        data: arguments.clone(),
350                    }],
351                    in_reply_to: None,
352                    expected_attempt: Some(self.task_attempt),
353                },
354            )
355            .await
356        {
357            Ok(stored) => stored,
358            Err(error) => return ToolExecutionResult::internal_error(error),
359        };
360        ToolExecutionResult::success(json!({
361            "status": "posted",
362            "task_id": self.task_id,
363            "message_id": stored.id,
364        }))
365    }
366
367    fn requires_context(&self) -> bool {
368        true
369    }
370}
371
372pub async fn report_result_tool_for_child_session(
373    child_session_id: SessionId,
374    session_store: &dyn SessionStore,
375    task_registry: &dyn crate::session_task::SessionTaskRegistry,
376) -> crate::error::Result<Option<ReportResultTool>> {
377    let Some((task, parent_workspace_id)) =
378        task_for_child_session(child_session_id, session_store, task_registry).await?
379    else {
380        return Ok(None);
381    };
382    let Some(schema) = declared_result_schema(&task).cloned() else {
383        return Ok(None);
384    };
385    Ok(Some(ReportResultTool::new(
386        task.session_id,
387        parent_workspace_id,
388        child_session_id,
389        task.id,
390        schema,
391    )))
392}
393
394pub async fn report_task_progress_tool_for_child_session(
395    child_session_id: SessionId,
396    session_store: &dyn SessionStore,
397    task_registry: &dyn crate::session_task::SessionTaskRegistry,
398) -> crate::error::Result<Option<ReportTaskProgressTool>> {
399    let Some((task, _)) =
400        task_for_child_session(child_session_id, session_store, task_registry).await?
401    else {
402        return Ok(None);
403    };
404    if task.state.is_terminal() {
405        return Ok(None);
406    }
407    let Some(schema) = declared_message_schema(&task).cloned() else {
408        return Ok(None);
409    };
410    Ok(Some(ReportTaskProgressTool::new(
411        task.session_id,
412        task.id,
413        task.attempt,
414        schema,
415    )))
416}
417
418pub(crate) async fn result_value_for_task(
419    context: &ToolContext,
420    task_id: Option<&str>,
421) -> Option<Value> {
422    let task_id = task_id?;
423    let registry = context.session_task_registry.as_ref()?;
424    let task = registry
425        .get(context.session_id, task_id)
426        .await
427        .ok()
428        .flatten()?;
429    declared_result_schema(&task)?;
430    let result_path = task.result_path.as_deref()?;
431    let file_store = context.file_store.as_ref()?;
432    let file = file_store
433        .read_file(context.workspace_fs_key(), result_path)
434        .await
435        .ok()
436        .flatten()?;
437    serde_json::from_str(file.content.as_deref()?).ok()
438}
439
440pub(crate) async fn required_result_is_missing(
441    context: &ToolContext,
442    task_id: Option<&str>,
443) -> bool {
444    let Some(task_id) = task_id else {
445        return false;
446    };
447    let Some(registry) = context.session_task_registry.as_ref() else {
448        return false;
449    };
450    registry
451        .get(context.session_id, task_id)
452        .await
453        .ok()
454        .flatten()
455        .is_some_and(|task| declared_result_schema(&task).is_some() && task.result_path.is_none())
456}
457
458pub(crate) async fn write_task_result_value(
459    context: &ToolContext,
460    task_id: &str,
461    value: &Value,
462) -> crate::error::Result<Option<String>> {
463    let Some(file_store) = context.file_store.as_ref() else {
464        return Ok(None);
465    };
466    let path = task_result_path(task_id);
467    let content = serde_json::to_string_pretty(value).map_err(|error| {
468        crate::error::AgentLoopError::store(format!("failed to serialize task result: {error}"))
469    })?;
470    file_store
471        .write_file(context.workspace_fs_key(), &path, &content, "utf-8")
472        .await?;
473    Ok(Some(path))
474}
475
476#[cfg(test)]
477mod tests {
478    use super::*;
479
480    #[test]
481    fn shared_validator_reports_required_type_and_extra_property_errors() {
482        let schema = json!({
483            "type": "object",
484            "properties": {
485                "answer": {"type": "string"},
486                "count": {"type": "integer"}
487            },
488            "required": ["answer", "count"],
489            "additionalProperties": false
490        });
491        let errors =
492            schema_validation_errors(&schema, &json!({"count": "not-an-integer", "extra": true}));
493        assert!(
494            errors
495                .iter()
496                .any(|error| error.contains("answer") && error.contains("required"))
497        );
498        assert!(
499            errors
500                .iter()
501                .any(|error| error.contains("count") && error.contains("integer"))
502        );
503        assert!(errors.iter().any(|error| error.contains("extra")
504            && (error.contains("additional") || error.contains("not allowed"))));
505    }
506
507    #[test]
508    fn shared_validator_enforces_full_json_schema_constraints() {
509        let schema = json!({
510            "type": "object",
511            "properties": {
512                "echo": {"type": "string", "pattern": "^[0-9]+$", "minLength": 2},
513                "score": {"type": "integer", "minimum": 0},
514                "tags": {
515                    "type": "array",
516                    "minItems": 2,
517                    "items": {"type": "string"}
518                },
519                "choice": {
520                    "oneOf": [
521                        {"const": "alpha"},
522                        {"const": "beta"}
523                    ]
524                }
525            },
526            "required": ["echo", "score", "tags", "choice"],
527            "additionalProperties": false
528        });
529
530        let errors = schema_validation_errors(
531            &schema,
532            &json!({
533                "echo": "x",
534                "score": -5,
535                "tags": ["solo"],
536                "choice": "gamma"
537            }),
538        );
539
540        assert!(errors.iter().any(|error| error.contains("echo")
541            && (error.contains("pattern") || error.contains("^[0-9]+$"))));
542        assert!(
543            errors
544                .iter()
545                .any(|error| error.contains("score") && error.contains("0"))
546        );
547        assert!(
548            errors
549                .iter()
550                .any(|error| error.contains("tags") && error.contains("2"))
551        );
552        assert!(
553            errors
554                .iter()
555                .any(|error| error.contains("choice") && error.contains("oneOf"))
556        );
557    }
558
559    #[test]
560    fn shared_schema_normalization_rejects_non_objects() {
561        let ToolExecutionResult::ToolError(error) =
562            normalize_result_schema(&json!({"result_schema": "object"})).unwrap_err()
563        else {
564            panic!("expected tool error");
565        };
566        assert!(error.contains("result_schema must be a JSON Schema object"));
567    }
568}
everruns_core/capabilities/delegation_result.rs

everruns_core/capabilities/
delegation_result.rs
