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 name(&self) -> &str {
147        "report_result"
148    }
149
150    fn display_name(&self) -> Option<&str> {
151        Some("Report Result")
152    }
153
154    fn description(&self) -> &str {
155        "Submit the final structured result for this delegated task. The call arguments must match the declared result schema."
156    }
157
158    fn parameters_schema(&self) -> Value {
159        self.result_schema.clone()
160    }
161
162    async fn execute(&self, _arguments: Value) -> ToolExecutionResult {
163        ToolExecutionResult::tool_error(
164            "report_result requires context. This tool must be executed with session context.",
165        )
166    }
167
168    async fn execute_with_context(
169        &self,
170        arguments: Value,
171        context: &ToolContext,
172    ) -> ToolExecutionResult {
173        let errors = schema_validation_errors(&self.result_schema, &arguments);
174        if !errors.is_empty() {
175            return ToolExecutionResult::tool_error(format!(
176                "report_result arguments do not match result_schema: {}",
177                errors.join("; ")
178            ));
179        }
180        let Some(registry) = context.session_task_registry.as_ref() else {
181            return ToolExecutionResult::tool_error(
182                "report_result requires session_task_registry context",
183            );
184        };
185        let Some(file_store) = self.file_store.as_ref().or(context.file_store.as_ref()) else {
186            return ToolExecutionResult::tool_error("report_result requires file_store context");
187        };
188
189        if context.session_id != self.child_session_id {
190            return ToolExecutionResult::tool_error(
191                "report_result can only be called from the linked child session",
192            );
193        }
194        let task = match registry.get(self.parent_session_id, &self.task_id).await {
195            Ok(Some(task)) => task,
196            Ok(None) => return ToolExecutionResult::tool_error("report_result task was not found"),
197            Err(error) => return ToolExecutionResult::internal_error(error),
198        };
199        if task.links.child_session_id != Some(self.child_session_id) {
200            return ToolExecutionResult::tool_error(
201                "report_result child session is no longer linked to this task",
202            );
203        }
204        if task.state.is_terminal() {
205            return ToolExecutionResult::tool_error(
206                "report_result is closed because the subagent task is terminal",
207            );
208        }
209        if task.result_path.is_some() {
210            return ToolExecutionResult::tool_error(
211                "report_result was already recorded for this subagent task",
212            );
213        }
214
215        let path = task_result_path(&self.task_id);
216        let content = match serde_json::to_string_pretty(&arguments) {
217            Ok(content) => content,
218            Err(error) => return ToolExecutionResult::internal_error(error),
219        };
220        let parent_workspace_key = SessionId::from_uuid(self.parent_workspace_id.uuid());
221        if let Err(error) = file_store
222            .write_file(parent_workspace_key, &path, &content, "utf-8")
223            .await
224        {
225            return ToolExecutionResult::internal_error(error);
226        }
227        if let Err(error) = registry
228            .update(
229                self.parent_session_id,
230                &self.task_id,
231                SessionTaskUpdate {
232                    // Fence result writes to the active attempt and repeat the
233                    // observed non-terminal state so a task that finalized after
234                    // our read rejects this stale content mutation too.
235                    expected_attempt: Some(task.attempt),
236                    state: Some(task.state),
237                    result_path: Some(path.clone()),
238                    summary: Some(truncate_summary(&content)),
239                    ..Default::default()
240                },
241            )
242            .await
243        {
244            return ToolExecutionResult::internal_error(error);
245        }
246        ToolExecutionResult::success(json!({
247            "status": "recorded",
248            "task_id": self.task_id,
249            "result_path": path,
250        }))
251    }
252
253    fn requires_context(&self) -> bool {
254        true
255    }
256}
257
258pub struct ReportTaskProgressTool {
259    parent_session_id: SessionId,
260    task_id: String,
261    task_attempt: i32,
262    message_schema: Value,
263}
264
265impl ReportTaskProgressTool {
266    pub fn new(
267        parent_session_id: SessionId,
268        task_id: String,
269        task_attempt: i32,
270        message_schema: Value,
271    ) -> Self {
272        Self {
273            parent_session_id,
274            task_id,
275            task_attempt,
276            message_schema,
277        }
278    }
279}
280
281#[async_trait]
282impl Tool for ReportTaskProgressTool {
283    fn name(&self) -> &str {
284        "report_task_progress"
285    }
286
287    fn display_name(&self) -> Option<&str> {
288        Some("Report Task Progress")
289    }
290
291    fn description(&self) -> &str {
292        "Post a structured progress message for this delegated task. The call arguments must match the declared message schema."
293    }
294
295    fn parameters_schema(&self) -> Value {
296        self.message_schema.clone()
297    }
298
299    async fn execute(&self, _arguments: Value) -> ToolExecutionResult {
300        ToolExecutionResult::tool_error(
301            "report_task_progress requires context. This tool must be executed with session context.",
302        )
303    }
304
305    async fn execute_with_context(
306        &self,
307        arguments: Value,
308        context: &ToolContext,
309    ) -> ToolExecutionResult {
310        let errors = schema_validation_errors(&self.message_schema, &arguments);
311        if !errors.is_empty() {
312            return ToolExecutionResult::tool_error(format!(
313                "report_task_progress arguments do not match message_schema: {}",
314                errors.join("; ")
315            ));
316        }
317        let Some(registry) = context.session_task_registry.as_ref() else {
318            return ToolExecutionResult::tool_error(
319                "report_task_progress requires session_task_registry context",
320            );
321        };
322        let stored = match registry
323            .record_message(
324                self.parent_session_id,
325                &self.task_id,
326                NewTaskMessage {
327                    direction: TaskMessageDirection::Outbound,
328                    content: vec![TaskMessagePart::Data {
329                        data: arguments.clone(),
330                    }],
331                    in_reply_to: None,
332                    expected_attempt: Some(self.task_attempt),
333                },
334            )
335            .await
336        {
337            Ok(stored) => stored,
338            Err(error) => return ToolExecutionResult::internal_error(error),
339        };
340        ToolExecutionResult::success(json!({
341            "status": "posted",
342            "task_id": self.task_id,
343            "message_id": stored.id,
344        }))
345    }
346
347    fn requires_context(&self) -> bool {
348        true
349    }
350}
351
352pub async fn report_result_tool_for_child_session(
353    child_session_id: SessionId,
354    session_store: &dyn SessionStore,
355    task_registry: &dyn crate::session_task::SessionTaskRegistry,
356) -> crate::error::Result<Option<ReportResultTool>> {
357    let Some((task, parent_workspace_id)) =
358        task_for_child_session(child_session_id, session_store, task_registry).await?
359    else {
360        return Ok(None);
361    };
362    let Some(schema) = declared_result_schema(&task).cloned() else {
363        return Ok(None);
364    };
365    Ok(Some(ReportResultTool::new(
366        task.session_id,
367        parent_workspace_id,
368        child_session_id,
369        task.id,
370        schema,
371    )))
372}
373
374pub async fn report_task_progress_tool_for_child_session(
375    child_session_id: SessionId,
376    session_store: &dyn SessionStore,
377    task_registry: &dyn crate::session_task::SessionTaskRegistry,
378) -> crate::error::Result<Option<ReportTaskProgressTool>> {
379    let Some((task, _)) =
380        task_for_child_session(child_session_id, session_store, task_registry).await?
381    else {
382        return Ok(None);
383    };
384    if task.state.is_terminal() {
385        return Ok(None);
386    }
387    let Some(schema) = declared_message_schema(&task).cloned() else {
388        return Ok(None);
389    };
390    Ok(Some(ReportTaskProgressTool::new(
391        task.session_id,
392        task.id,
393        task.attempt,
394        schema,
395    )))
396}
397
398pub(crate) async fn result_value_for_task(
399    context: &ToolContext,
400    task_id: Option<&str>,
401) -> Option<Value> {
402    let task_id = task_id?;
403    let registry = context.session_task_registry.as_ref()?;
404    let task = registry
405        .get(context.session_id, task_id)
406        .await
407        .ok()
408        .flatten()?;
409    declared_result_schema(&task)?;
410    let result_path = task.result_path.as_deref()?;
411    let file_store = context.file_store.as_ref()?;
412    let file = file_store
413        .read_file(context.workspace_fs_key(), result_path)
414        .await
415        .ok()
416        .flatten()?;
417    serde_json::from_str(file.content.as_deref()?).ok()
418}
419
420pub(crate) async fn required_result_is_missing(
421    context: &ToolContext,
422    task_id: Option<&str>,
423) -> bool {
424    let Some(task_id) = task_id else {
425        return false;
426    };
427    let Some(registry) = context.session_task_registry.as_ref() else {
428        return false;
429    };
430    registry
431        .get(context.session_id, task_id)
432        .await
433        .ok()
434        .flatten()
435        .is_some_and(|task| declared_result_schema(&task).is_some() && task.result_path.is_none())
436}
437
438pub(crate) async fn write_task_result_value(
439    context: &ToolContext,
440    task_id: &str,
441    value: &Value,
442) -> crate::error::Result<Option<String>> {
443    let Some(file_store) = context.file_store.as_ref() else {
444        return Ok(None);
445    };
446    let path = task_result_path(task_id);
447    let content = serde_json::to_string_pretty(value).map_err(|error| {
448        crate::error::AgentLoopError::store(format!("failed to serialize task result: {error}"))
449    })?;
450    file_store
451        .write_file(context.workspace_fs_key(), &path, &content, "utf-8")
452        .await?;
453    Ok(Some(path))
454}
455
456#[cfg(test)]
457mod tests {
458    use super::*;
459
460    #[test]
461    fn shared_validator_reports_required_type_and_extra_property_errors() {
462        let schema = json!({
463            "type": "object",
464            "properties": {
465                "answer": {"type": "string"},
466                "count": {"type": "integer"}
467            },
468            "required": ["answer", "count"],
469            "additionalProperties": false
470        });
471        let errors =
472            schema_validation_errors(&schema, &json!({"count": "not-an-integer", "extra": true}));
473        assert!(
474            errors
475                .iter()
476                .any(|error| error.contains("answer") && error.contains("required"))
477        );
478        assert!(
479            errors
480                .iter()
481                .any(|error| error.contains("count") && error.contains("integer"))
482        );
483        assert!(errors.iter().any(|error| error.contains("extra")
484            && (error.contains("additional") || error.contains("not allowed"))));
485    }
486
487    #[test]
488    fn shared_validator_enforces_full_json_schema_constraints() {
489        let schema = json!({
490            "type": "object",
491            "properties": {
492                "echo": {"type": "string", "pattern": "^[0-9]+$", "minLength": 2},
493                "score": {"type": "integer", "minimum": 0},
494                "tags": {
495                    "type": "array",
496                    "minItems": 2,
497                    "items": {"type": "string"}
498                },
499                "choice": {
500                    "oneOf": [
501                        {"const": "alpha"},
502                        {"const": "beta"}
503                    ]
504                }
505            },
506            "required": ["echo", "score", "tags", "choice"],
507            "additionalProperties": false
508        });
509
510        let errors = schema_validation_errors(
511            &schema,
512            &json!({
513                "echo": "x",
514                "score": -5,
515                "tags": ["solo"],
516                "choice": "gamma"
517            }),
518        );
519
520        assert!(errors.iter().any(|error| error.contains("echo")
521            && (error.contains("pattern") || error.contains("^[0-9]+$"))));
522        assert!(
523            errors
524                .iter()
525                .any(|error| error.contains("score") && error.contains("0"))
526        );
527        assert!(
528            errors
529                .iter()
530                .any(|error| error.contains("tags") && error.contains("2"))
531        );
532        assert!(
533            errors
534                .iter()
535                .any(|error| error.contains("choice") && error.contains("oneOf"))
536        );
537    }
538
539    #[test]
540    fn shared_schema_normalization_rejects_non_objects() {
541        let ToolExecutionResult::ToolError(error) =
542            normalize_result_schema(&json!({"result_schema": "object"})).unwrap_err()
543        else {
544            panic!("expected tool error");
545        };
546        assert!(error.contains("result_schema must be a JSON Schema object"));
547    }
548}
everruns_core/capabilities/delegation_result.rs

everruns_core/capabilities/
delegation_result.rs
