mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
feat: structured fallback deliverables for failed/stuck jobs (#236)
* feat: structured fallback deliverables for failed/stuck jobs (#221) When a job fails or gets stuck, build a FallbackDeliverable that captures partial results, action statistics, cost, timing, and repair attempts. This replaces opaque error strings with structured data users can act on. - Add FallbackDeliverable, LastAction, ActionStats types in context/fallback.rs - Store fallback in JobContext.metadata["fallback_deliverable"] on failure - Surface fallback in job_status tool output and SSE job_result events - Update mark_failed() and mark_stuck() in worker to build fallback - 8 unit tests covering zero/mixed actions, truncation, timing, serialization Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address review comments on fallback deliverables - Fix doc comment: "200 chars" -> "200 bytes (UTF-8 safe)" since truncate_str operates on byte length, not character count. - Add code comment documenting that SSE fallback_deliverable is currently always None (forward-compatible infrastructure). Co-Authored-By: Claude Opus 4.6 <[email protected]> * refactor: take Option<&FallbackDeliverable> instead of &Option<…> Addresses Gemini review feedback: idiomatic Rust prefers Option<&T> over &Option<T> for borrowed optional values. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: guard against non-object metadata and add fallback test - store_fallback_in_metadata now resets metadata to {} when it's any non-object type (string, array, number), not just null. Prevents panic on index assignment. - Add test_job_status_includes_fallback_deliverable to verify the fallback field is surfaced in job_status tool output. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: use sanitized output in fallback preview + add integration tests Security fix: FallbackDeliverable::build() now uses output_sanitized instead of output_raw, preventing secrets/PII from leaking through the job_status tool and SSE job_result events. Also adds: - test_fallback_uses_sanitized_output: proves raw secrets don't leak - test_store_fallback_in_metadata_roundtrip: full serialize/deserialize - test_store_fallback_handles_non_object_metadata: edge case coverage - test_store_fallback_none_is_noop: None input is safe Addresses serrrfirat review feedback on PR #236. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: harden fallback deliverables against review findings - Truncate failure_reason to 1000 bytes to prevent metadata bloat - Add tracing::warn on fallback serialization failure (was silently discarded) - Fix module/struct docs to cover stuck jobs, remove stale SSE claim - Fix job.rs test to use real FallbackDeliverable field names - Add tests for failure_reason truncation and completed_at=None elapsed time - Fix pre-existing clippy warning in settings.rs (field_reassign_with_default) Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address Copilot review findings on fallback deliverables - Fix output_raw/output_sanitized field swap in ActionRecord::succeed() so sanitized data actually goes into the sanitized field (security) - Return None instead of empty Memory when get_memory fails in build_fallback, with tracing::warn for observability - Replace manual elapsed calculation with ctx.elapsed() which already clamps negative durations Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: resolve rebase conflicts and update tests for parameter swap - Add fallback field to SseEvent::JobResult in job_monitor - Fix type annotation in fallback deliverable test - Update test_action_record_succeed_sets_fields for new parameter order - Use create_job_for_user in test (API changed on main) Co-Authored-By: Claude Opus 4.6 <[email protected]> * chore: trigger CI re-check after rebase * fix: fall back to error message for failed action output_preview When the last action is a failed tool call, output_sanitized is None, leaving output_preview empty. Now falls back to the action's error message so users see what went wrong. [skip-regression-check] * ci: add safety comments to test code for no-panics check The CI no-panics grep check cannot distinguish test code inside src/ files from production code. Add // safety: test annotations to .unwrap(), .expect(), and assert!() calls in #[cfg(test)] modules. * fix: clarify succeed() doc and avoid clone in output_preview - Fix doc comment: output_raw is stored as pretty-printed JSON string, not a raw JSON value - Borrow string slice directly in fallback preview to avoid cloning potentially large sanitized outputs before truncation * refactor: reuse floor_char_boundary in truncate_str Replace hand-rolled UTF-8 boundary logic with existing crate::util::floor_char_boundary to reduce duplication. * fix: rename SSE fallback field to fallback_deliverable for consistency The SSE JobResult field was named `fallback` while everywhere else (metadata key, job_status tool) uses `fallback_deliverable`. Align the SSE wire format to avoid forcing clients to handle two names. --------- Co-authored-by: Claude Opus 4.6 <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
86ae12747b
commit
65062f3cc0
@@ -211,6 +211,7 @@ mod tests {
|
||||
job_id: job_id.to_string(),
|
||||
status: "completed".to_string(),
|
||||
session_id: None,
|
||||
fallback_deliverable: None,
|
||||
},
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
@@ -232,6 +232,8 @@ pub enum SseEvent {
|
||||
status: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
session_id: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
fallback_deliverable: Option<serde_json::Value>,
|
||||
},
|
||||
|
||||
/// An image was generated by a tool.
|
||||
|
||||
@@ -0,0 +1,319 @@
|
||||
//! Structured fallback deliverables for failed or stuck jobs.
|
||||
//!
|
||||
//! When a job fails or is detected as stuck, a [`FallbackDeliverable`] captures
|
||||
//! what was accomplished before the failure: partial results, action statistics,
|
||||
//! cost, and timing. This gives users visibility into terminal jobs instead of
|
||||
//! just an error string.
|
||||
//!
|
||||
//! Fallback deliverables are stored in `JobContext.metadata["fallback_deliverable"]`
|
||||
//! and surfaced through the `job_status` tool.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::context::memory::Memory;
|
||||
use crate::context::state::JobContext;
|
||||
|
||||
/// Structured summary of a failed or stuck job.
|
||||
///
|
||||
/// Stored in `JobContext.metadata["fallback_deliverable"]` when a job fails
|
||||
/// or is marked stuck. Surfaced through the `job_status` tool.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct FallbackDeliverable {
|
||||
/// True if at least one action succeeded before failure.
|
||||
pub partial: bool,
|
||||
/// Why the job failed.
|
||||
pub failure_reason: String,
|
||||
/// Last action taken before failure.
|
||||
pub last_action: Option<LastAction>,
|
||||
/// Aggregate action statistics.
|
||||
pub action_stats: ActionStats,
|
||||
/// Total tokens consumed.
|
||||
pub tokens_used: u64,
|
||||
/// Total cost incurred (decimal as string for JSON safety).
|
||||
pub cost: String,
|
||||
/// Wall-clock elapsed time in seconds.
|
||||
pub elapsed_secs: f64,
|
||||
/// Number of self-repair attempts.
|
||||
pub repair_attempts: u32,
|
||||
}
|
||||
|
||||
/// Summary of the last action taken before failure.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct LastAction {
|
||||
pub tool_name: String,
|
||||
/// Truncated to 200 bytes (UTF-8 safe).
|
||||
pub output_preview: String,
|
||||
pub success: bool,
|
||||
}
|
||||
|
||||
/// Aggregate action counts.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ActionStats {
|
||||
pub total: u32,
|
||||
pub successful: u32,
|
||||
pub failed: u32,
|
||||
}
|
||||
|
||||
impl FallbackDeliverable {
|
||||
/// Build a fallback deliverable from a job context and its memory.
|
||||
pub fn build(ctx: &JobContext, memory: &Memory, reason: &str) -> Self {
|
||||
let successful = memory.successful_actions() as u32;
|
||||
let failed = memory.failed_actions() as u32;
|
||||
let total = memory.actions.len() as u32;
|
||||
|
||||
let last_action = memory.last_action().map(|a| {
|
||||
// Use sanitized output to avoid leaking secrets through the fallback API surface.
|
||||
// For failed actions (no sanitized output), fall back to the error message.
|
||||
// Borrow the string slice directly when possible to avoid cloning
|
||||
// potentially large outputs just for truncation.
|
||||
let owned_fallback;
|
||||
let preview_str: &str = if let Some(v) = a.output_sanitized.as_ref() {
|
||||
match v {
|
||||
serde_json::Value::String(s) => s.as_str(),
|
||||
other => {
|
||||
owned_fallback = serde_json::to_string(other).unwrap_or_default();
|
||||
&owned_fallback
|
||||
}
|
||||
}
|
||||
} else if let Some(ref err) = a.error {
|
||||
err.as_str()
|
||||
} else {
|
||||
""
|
||||
};
|
||||
let preview = truncate_str(preview_str, 200);
|
||||
LastAction {
|
||||
tool_name: a.tool_name.clone(),
|
||||
output_preview: preview.to_string(),
|
||||
success: a.success,
|
||||
}
|
||||
});
|
||||
|
||||
let elapsed_secs = ctx.elapsed().map_or(0.0, |d| d.as_secs_f64());
|
||||
|
||||
Self {
|
||||
partial: successful > 0,
|
||||
failure_reason: truncate_str(reason, 1000).to_string(),
|
||||
last_action,
|
||||
action_stats: ActionStats {
|
||||
total,
|
||||
successful,
|
||||
failed,
|
||||
},
|
||||
tokens_used: ctx.total_tokens_used,
|
||||
cost: ctx.actual_cost.to_string(),
|
||||
elapsed_secs,
|
||||
repair_attempts: ctx.repair_attempts,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Truncate a string to at most `max_len` bytes on a char boundary.
|
||||
fn truncate_str(s: &str, max_len: usize) -> &str {
|
||||
&s[..crate::util::floor_char_boundary(s, max_len)]
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::context::memory::Memory;
|
||||
use crate::context::state::JobContext;
|
||||
use chrono::{Duration, Utc};
|
||||
use rust_decimal::Decimal;
|
||||
use std::time::Duration as StdDuration;
|
||||
|
||||
#[test]
|
||||
fn test_fallback_zero_actions() {
|
||||
let ctx = JobContext::new("Test", "Empty job");
|
||||
let memory = Memory::new(ctx.job_id);
|
||||
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "timed out");
|
||||
|
||||
assert!(!fb.partial); // safety: test
|
||||
assert_eq!(fb.failure_reason, "timed out"); // safety: test
|
||||
assert!(fb.last_action.is_none()); // safety: test
|
||||
assert_eq!(fb.action_stats.total, 0); // safety: test
|
||||
assert_eq!(fb.action_stats.successful, 0); // safety: test
|
||||
assert_eq!(fb.action_stats.failed, 0); // safety: test
|
||||
assert_eq!(fb.tokens_used, 0); // safety: test
|
||||
assert_eq!(fb.cost, "0"); // safety: test
|
||||
assert_eq!(fb.repair_attempts, 0); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_mixed_actions() {
|
||||
let mut ctx = JobContext::new("Test", "Mixed job");
|
||||
ctx.total_tokens_used = 5000;
|
||||
ctx.actual_cost = Decimal::new(42, 2); // 0.42
|
||||
ctx.repair_attempts = 1;
|
||||
|
||||
let mut memory = Memory::new(ctx.job_id);
|
||||
|
||||
// 3 successes
|
||||
for _ in 0..3 {
|
||||
let action = memory
|
||||
.create_action("tool_a", serde_json::json!({}))
|
||||
.succeed(
|
||||
Some("output".to_string()),
|
||||
serde_json::json!({}),
|
||||
StdDuration::from_secs(1),
|
||||
);
|
||||
memory.record_action(action);
|
||||
}
|
||||
// 2 failures
|
||||
for _ in 0..2 {
|
||||
let action = memory
|
||||
.create_action("tool_b", serde_json::json!({}))
|
||||
.fail("broke", StdDuration::from_secs(1));
|
||||
memory.record_action(action);
|
||||
}
|
||||
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "max iterations");
|
||||
|
||||
assert!(fb.partial); // safety: test
|
||||
assert_eq!(fb.action_stats.total, 5); // safety: test
|
||||
assert_eq!(fb.action_stats.successful, 3); // safety: test
|
||||
assert_eq!(fb.action_stats.failed, 2); // safety: test
|
||||
assert_eq!(fb.tokens_used, 5000); // safety: test
|
||||
assert_eq!(fb.cost, "0.42"); // safety: test
|
||||
assert_eq!(fb.repair_attempts, 1); // safety: test
|
||||
assert!(fb.last_action.is_some()); // safety: test
|
||||
let la = fb.last_action.unwrap(); // safety: test
|
||||
assert_eq!(la.tool_name, "tool_b"); // safety: test
|
||||
assert!(!la.success); // safety: test
|
||||
// Failed actions should surface the error message as the output preview
|
||||
assert_eq!(la.output_preview, "broke"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_failed_action_shows_error() {
|
||||
let ctx = JobContext::new("Test", "Error preview");
|
||||
let mut memory = Memory::new(ctx.job_id);
|
||||
|
||||
let action = memory
|
||||
.create_action("broken_tool", serde_json::json!({}))
|
||||
.fail("connection timed out after 30s", StdDuration::from_secs(30));
|
||||
memory.record_action(action);
|
||||
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "tool failure");
|
||||
let la = fb.last_action.unwrap(); // safety: test
|
||||
assert!(!la.success); // safety: test
|
||||
assert_eq!(la.output_preview, "connection timed out after 30s"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_last_action_truncation() {
|
||||
let ctx = JobContext::new("Test", "Truncation");
|
||||
let mut memory = Memory::new(ctx.job_id);
|
||||
|
||||
let long_output = "x".repeat(500);
|
||||
let action = memory
|
||||
.create_action("tool_c", serde_json::json!({}))
|
||||
.succeed(
|
||||
Some(long_output.clone()),
|
||||
serde_json::Value::String(long_output),
|
||||
StdDuration::from_secs(1),
|
||||
);
|
||||
memory.record_action(action);
|
||||
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "failed");
|
||||
let la = fb.last_action.unwrap(); // safety: test
|
||||
assert!(la.output_preview.len() <= 200); // safety: test
|
||||
assert!(!la.output_preview.is_empty()); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_uses_sanitized_output() {
|
||||
let ctx = JobContext::new("Test", "Sanitized");
|
||||
let mut memory = Memory::new(ctx.job_id);
|
||||
|
||||
let action = memory
|
||||
.create_action("tool_d", serde_json::json!({}))
|
||||
.succeed(
|
||||
Some("[REDACTED]".to_string()),
|
||||
serde_json::json!({"api_key": "sk-secret-key-12345"}),
|
||||
StdDuration::from_secs(1),
|
||||
);
|
||||
memory.record_action(action);
|
||||
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "failed");
|
||||
let la = fb.last_action.unwrap(); // safety: test
|
||||
// Must use sanitized output, not raw
|
||||
assert!(!la.output_preview.contains("sk-secret")); // safety: test
|
||||
assert!(la.output_preview.contains("REDACTED")); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_elapsed_time() {
|
||||
let mut ctx = JobContext::new("Test", "Timing");
|
||||
let now = Utc::now();
|
||||
ctx.started_at = Some(now - Duration::seconds(10));
|
||||
ctx.completed_at = Some(now);
|
||||
|
||||
let memory = Memory::new(ctx.job_id);
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "failed");
|
||||
|
||||
// Should be approximately 10 seconds
|
||||
assert!((fb.elapsed_secs - 10.0).abs() < 0.1); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_no_started_at() {
|
||||
let ctx = JobContext::new("Test", "Never started");
|
||||
let memory = Memory::new(ctx.job_id);
|
||||
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "failed");
|
||||
assert!((fb.elapsed_secs - 0.0).abs() < 0.001); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_elapsed_time_no_completed_at() {
|
||||
let mut ctx = JobContext::new("Test", "Still running");
|
||||
ctx.started_at = Some(Utc::now() - Duration::seconds(5));
|
||||
// completed_at is None — should use Utc::now() as fallback
|
||||
|
||||
let memory = Memory::new(ctx.job_id);
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "stuck");
|
||||
|
||||
// Should be approximately 5 seconds (using now as end time)
|
||||
assert!(fb.elapsed_secs >= 4.0 && fb.elapsed_secs <= 7.0); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_failure_reason_truncation() {
|
||||
let ctx = JobContext::new("Test", "Long reason");
|
||||
let memory = Memory::new(ctx.job_id);
|
||||
|
||||
let long_reason = "x".repeat(5000);
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, &long_reason);
|
||||
|
||||
assert!(fb.failure_reason.len() <= 1000); // safety: test
|
||||
assert!(!fb.failure_reason.is_empty()); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_str_ascii() {
|
||||
assert_eq!(truncate_str("hello", 10), "hello"); // safety: test
|
||||
assert_eq!(truncate_str("hello world", 5), "hello"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_str_unicode() {
|
||||
// "é" is 2 bytes in UTF-8
|
||||
let s = "café";
|
||||
assert_eq!(truncate_str(s, 10), "café"); // safety: test
|
||||
// Truncating at 4 would split "é", should back up to 3
|
||||
assert_eq!(truncate_str(s, 4), "caf"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_fallback_serialization() {
|
||||
let ctx = JobContext::new("Test", "Serialize");
|
||||
let memory = Memory::new(ctx.job_id);
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "test error");
|
||||
|
||||
// Should serialize to JSON and back without error
|
||||
let json = serde_json::to_value(&fb).unwrap(); // safety: test
|
||||
let deserialized: FallbackDeliverable = serde_json::from_value(json).unwrap(); // safety: test
|
||||
assert_eq!(deserialized.failure_reason, "test error"); // safety: test
|
||||
}
|
||||
}
|
||||
+79
-70
@@ -58,15 +58,19 @@ impl ActionRecord {
|
||||
}
|
||||
|
||||
/// Mark the action as successful.
|
||||
///
|
||||
/// `output_sanitized` is the tool output after safety processing (string).
|
||||
/// `output_raw` is the original tool result (JSON value, stored as a
|
||||
/// pretty-printed JSON string in `ActionRecord.output_raw`).
|
||||
pub fn succeed(
|
||||
mut self,
|
||||
output_raw: Option<String>,
|
||||
output_sanitized: serde_json::Value,
|
||||
output_sanitized: Option<String>,
|
||||
output_raw: serde_json::Value,
|
||||
duration: Duration,
|
||||
) -> Self {
|
||||
self.success = true;
|
||||
self.output_raw = output_raw;
|
||||
self.output_sanitized = Some(output_sanitized);
|
||||
self.output_raw = Some(serde_json::to_string_pretty(&output_raw).unwrap_or_default());
|
||||
self.output_sanitized = output_sanitized.map(serde_json::Value::String);
|
||||
self.duration = duration;
|
||||
self
|
||||
}
|
||||
@@ -248,15 +252,15 @@ mod tests {
|
||||
#[test]
|
||||
fn test_action_record() {
|
||||
let action = ActionRecord::new(0, "test", serde_json::json!({"key": "value"}));
|
||||
assert_eq!(action.sequence, 0);
|
||||
assert!(!action.success);
|
||||
assert_eq!(action.sequence, 0); // safety: test
|
||||
assert!(!action.success); // safety: test
|
||||
|
||||
let action = action.succeed(
|
||||
Some("raw".to_string()),
|
||||
serde_json::json!({"result": "ok"}),
|
||||
Duration::from_millis(100),
|
||||
);
|
||||
assert!(action.success);
|
||||
assert!(action.success); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -267,7 +271,7 @@ mod tests {
|
||||
memory.add(ChatMessage::user("How are you?"));
|
||||
memory.add(ChatMessage::assistant("Good!"));
|
||||
|
||||
assert_eq!(memory.len(), 3); // Oldest removed
|
||||
assert_eq!(memory.len(), 3); // Oldest removed // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -286,9 +290,9 @@ mod tests {
|
||||
.with_cost(Decimal::new(20, 1));
|
||||
memory.record_action(action2);
|
||||
|
||||
assert_eq!(memory.total_cost(), Decimal::new(30, 1));
|
||||
assert_eq!(memory.total_duration(), Duration::from_secs(3));
|
||||
assert_eq!(memory.successful_actions(), 2);
|
||||
assert_eq!(memory.total_cost(), Decimal::new(30, 1)); // safety: test
|
||||
assert_eq!(memory.total_duration(), Duration::from_secs(3)); // safety: test
|
||||
assert_eq!(memory.successful_actions(), 2); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -296,11 +300,11 @@ mod tests {
|
||||
let action = ActionRecord::new(1, "broken_tool", serde_json::json!({"x": 1}));
|
||||
let action = action.fail("something went wrong", Duration::from_millis(50));
|
||||
|
||||
assert!(!action.success);
|
||||
assert_eq!(action.error.as_deref(), Some("something went wrong"));
|
||||
assert_eq!(action.duration, Duration::from_millis(50));
|
||||
assert!(action.output_raw.is_none());
|
||||
assert!(action.output_sanitized.is_none());
|
||||
assert!(!action.success); // safety: test
|
||||
assert_eq!(action.error.as_deref(), Some("something went wrong")); // safety: test
|
||||
assert_eq!(action.duration, Duration::from_millis(50)); // safety: test
|
||||
assert!(action.output_raw.is_none()); // safety: test
|
||||
assert!(action.output_sanitized.is_none()); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -308,9 +312,9 @@ mod tests {
|
||||
let action = ActionRecord::new(0, "risky_tool", serde_json::json!({}));
|
||||
let action = action.with_warnings(vec!["suspicious pattern".into(), "possible xss".into()]);
|
||||
|
||||
assert_eq!(action.sanitization_warnings.len(), 2);
|
||||
assert_eq!(action.sanitization_warnings[0], "suspicious pattern");
|
||||
assert_eq!(action.sanitization_warnings[1], "possible xss");
|
||||
assert_eq!(action.sanitization_warnings.len(), 2); // safety: test
|
||||
assert_eq!(action.sanitization_warnings[0], "suspicious pattern"); // safety: test
|
||||
assert_eq!(action.sanitization_warnings[1], "possible xss"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -319,41 +323,46 @@ mod tests {
|
||||
let cost = Decimal::new(42, 2); // 0.42
|
||||
let action = action.with_cost(cost);
|
||||
|
||||
assert_eq!(action.cost, Some(Decimal::new(42, 2)));
|
||||
assert_eq!(action.cost, Some(Decimal::new(42, 2))); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_action_record_new_defaults() {
|
||||
let action = ActionRecord::new(5, "my_tool", serde_json::json!({"key": "val"}));
|
||||
|
||||
assert_eq!(action.sequence, 5);
|
||||
assert_eq!(action.tool_name, "my_tool");
|
||||
assert_eq!(action.input, serde_json::json!({"key": "val"}));
|
||||
assert!(!action.success);
|
||||
assert!(action.output_raw.is_none());
|
||||
assert!(action.output_sanitized.is_none());
|
||||
assert!(action.sanitization_warnings.is_empty());
|
||||
assert!(action.cost.is_none());
|
||||
assert_eq!(action.duration, Duration::ZERO);
|
||||
assert!(action.error.is_none());
|
||||
assert_eq!(action.sequence, 5); // safety: test
|
||||
assert_eq!(action.tool_name, "my_tool"); // safety: test
|
||||
assert_eq!(action.input, serde_json::json!({"key": "val"})); // safety: test
|
||||
assert!(!action.success); // safety: test
|
||||
assert!(action.output_raw.is_none()); // safety: test
|
||||
assert!(action.output_sanitized.is_none()); // safety: test
|
||||
assert!(action.sanitization_warnings.is_empty()); // safety: test
|
||||
assert!(action.cost.is_none()); // safety: test
|
||||
assert_eq!(action.duration, Duration::ZERO); // safety: test
|
||||
assert!(action.error.is_none()); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_action_record_succeed_sets_fields() {
|
||||
let action = ActionRecord::new(0, "tool", serde_json::json!({}));
|
||||
let action = action.succeed(
|
||||
Some("raw output here".into()),
|
||||
Some("sanitized output".into()),
|
||||
serde_json::json!({"clean": true}),
|
||||
Duration::from_secs(7),
|
||||
);
|
||||
|
||||
assert!(action.success);
|
||||
assert_eq!(action.output_raw.as_deref(), Some("raw output here"));
|
||||
assert!(action.success); // safety: test
|
||||
// output_raw is the JSON value pretty-printed
|
||||
let expected_raw =
|
||||
serde_json::to_string_pretty(&serde_json::json!({"clean": true})).unwrap(); // safety: test
|
||||
assert_eq!(action.output_raw.as_deref(), Some(expected_raw.as_str())); // safety: test
|
||||
// output_sanitized wraps the string in a JSON string value
|
||||
assert_eq!(
|
||||
/* safety: test */
|
||||
action.output_sanitized,
|
||||
Some(serde_json::json!({"clean": true}))
|
||||
Some(serde_json::json!("sanitized output"))
|
||||
);
|
||||
assert_eq!(action.duration, Duration::from_secs(7));
|
||||
assert_eq!(action.duration, Duration::from_secs(7)); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -361,13 +370,13 @@ mod tests {
|
||||
let mut mem = ConversationMemory::new(10);
|
||||
mem.add(ChatMessage::user("hello"));
|
||||
mem.add(ChatMessage::assistant("hi"));
|
||||
assert_eq!(mem.len(), 2);
|
||||
assert!(!mem.is_empty());
|
||||
assert_eq!(mem.len(), 2); // safety: test
|
||||
assert!(!mem.is_empty()); // safety: test
|
||||
|
||||
mem.clear();
|
||||
assert_eq!(mem.len(), 0);
|
||||
assert!(mem.is_empty());
|
||||
assert!(mem.messages().is_empty());
|
||||
assert_eq!(mem.len(), 0); // safety: test
|
||||
assert!(mem.is_empty()); // safety: test
|
||||
assert!(mem.messages().is_empty()); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -379,20 +388,20 @@ mod tests {
|
||||
mem.add(ChatMessage::assistant("four"));
|
||||
|
||||
let last_2 = mem.last_n(2);
|
||||
assert_eq!(last_2.len(), 2);
|
||||
assert_eq!(last_2[0].content, "three");
|
||||
assert_eq!(last_2[1].content, "four");
|
||||
assert_eq!(last_2.len(), 2); // safety: test
|
||||
assert_eq!(last_2[0].content, "three"); // safety: test
|
||||
assert_eq!(last_2[1].content, "four"); // safety: test
|
||||
|
||||
// Requesting more than available returns all
|
||||
let last_100 = mem.last_n(100);
|
||||
assert_eq!(last_100.len(), 4);
|
||||
assert_eq!(last_100.len(), 4); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_conversation_memory_last_n_empty() {
|
||||
let mem = ConversationMemory::new(10);
|
||||
let result = mem.last_n(5);
|
||||
assert!(result.is_empty());
|
||||
assert!(result.is_empty()); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -405,13 +414,13 @@ mod tests {
|
||||
// At capacity (3). Adding one more should trim, but keep system.
|
||||
mem.add(ChatMessage::user("msg3"));
|
||||
|
||||
assert_eq!(mem.len(), 3);
|
||||
assert_eq!(mem.len(), 3); // safety: test
|
||||
// System message must survive
|
||||
assert_eq!(mem.messages()[0].role, crate::llm::Role::System);
|
||||
assert_eq!(mem.messages()[0].content, "You are helpful");
|
||||
assert_eq!(mem.messages()[0].role, crate::llm::Role::System); // safety: test
|
||||
assert_eq!(mem.messages()[0].content, "You are helpful"); // safety: test
|
||||
// Oldest non-system message (msg1) should be gone
|
||||
assert_eq!(mem.messages()[1].content, "msg2");
|
||||
assert_eq!(mem.messages()[2].content, "msg3");
|
||||
assert_eq!(mem.messages()[1].content, "msg2"); // safety: test
|
||||
assert_eq!(mem.messages()[2].content, "msg3"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -422,9 +431,9 @@ mod tests {
|
||||
// Now at capacity. Add another.
|
||||
mem.add(ChatMessage::user("b"));
|
||||
|
||||
assert_eq!(mem.len(), 2);
|
||||
assert_eq!(mem.messages()[0].role, crate::llm::Role::System);
|
||||
assert_eq!(mem.messages()[1].content, "b");
|
||||
assert_eq!(mem.len(), 2); // safety: test
|
||||
assert_eq!(mem.messages()[0].role, crate::llm::Role::System); // safety: test
|
||||
assert_eq!(mem.messages()[1].content, "b"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -440,7 +449,7 @@ mod tests {
|
||||
mem.add(ChatMessage::user("hello"));
|
||||
// Should have broken out rather than looping forever.
|
||||
// The system message is protected, so len may exceed max.
|
||||
assert!(mem.len() <= 2);
|
||||
assert!(mem.len() <= 2); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -459,14 +468,14 @@ mod tests {
|
||||
.fail("oops", Duration::from_millis(2));
|
||||
memory.record_action(err);
|
||||
|
||||
assert_eq!(memory.successful_actions(), 1);
|
||||
assert_eq!(memory.failed_actions(), 1);
|
||||
assert_eq!(memory.successful_actions(), 1); // safety: test
|
||||
assert_eq!(memory.failed_actions(), 1); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_memory_last_action() {
|
||||
let mut memory = Memory::new(Uuid::new_v4());
|
||||
assert!(memory.last_action().is_none());
|
||||
assert!(memory.last_action().is_none()); // safety: test
|
||||
|
||||
let a1 = memory
|
||||
.create_action("first", serde_json::json!({}))
|
||||
@@ -478,8 +487,8 @@ mod tests {
|
||||
.fail("nope", Duration::ZERO);
|
||||
memory.record_action(a2);
|
||||
|
||||
let last = memory.last_action().unwrap();
|
||||
assert_eq!(last.tool_name, "second");
|
||||
let last = memory.last_action().unwrap(); // safety: test
|
||||
assert_eq!(last.tool_name, "second"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -499,9 +508,9 @@ mod tests {
|
||||
);
|
||||
memory.record_action(a);
|
||||
|
||||
assert_eq!(memory.actions_by_tool("shell").len(), 3);
|
||||
assert_eq!(memory.actions_by_tool("http").len(), 1);
|
||||
assert_eq!(memory.actions_by_tool("nonexistent").len(), 0);
|
||||
assert_eq!(memory.actions_by_tool("shell").len(), 3); // safety: test
|
||||
assert_eq!(memory.actions_by_tool("http").len(), 1); // safety: test
|
||||
assert_eq!(memory.actions_by_tool("nonexistent").len(), 0); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -509,25 +518,25 @@ mod tests {
|
||||
let mut memory = Memory::new(Uuid::new_v4());
|
||||
|
||||
let a0 = memory.create_action("t", serde_json::json!({}));
|
||||
assert_eq!(a0.sequence, 0);
|
||||
assert_eq!(a0.sequence, 0); // safety: test
|
||||
|
||||
let a1 = memory.create_action("t", serde_json::json!({}));
|
||||
assert_eq!(a1.sequence, 1);
|
||||
assert_eq!(a1.sequence, 1); // safety: test
|
||||
|
||||
let a2 = memory.create_action("t", serde_json::json!({}));
|
||||
assert_eq!(a2.sequence, 2);
|
||||
assert_eq!(a2.sequence, 2); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_memory_add_message_delegates_to_conversation() {
|
||||
let mut memory = Memory::new(Uuid::new_v4());
|
||||
assert!(memory.conversation.is_empty());
|
||||
assert!(memory.conversation.is_empty()); // safety: test
|
||||
|
||||
memory.add_message(ChatMessage::user("hello"));
|
||||
memory.add_message(ChatMessage::assistant("hi"));
|
||||
|
||||
assert_eq!(memory.conversation.len(), 2);
|
||||
assert_eq!(memory.conversation.messages()[0].content, "hello");
|
||||
assert_eq!(memory.conversation.len(), 2); // safety: test
|
||||
assert_eq!(memory.conversation.messages()[0].content, "hello"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -540,7 +549,7 @@ mod tests {
|
||||
.succeed(None, serde_json::json!({}), Duration::ZERO);
|
||||
memory.record_action(a);
|
||||
|
||||
assert_eq!(memory.total_cost(), Decimal::ZERO);
|
||||
assert_eq!(memory.total_cost(), Decimal::ZERO); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -560,6 +569,6 @@ mod tests {
|
||||
memory.record_action(a2);
|
||||
|
||||
// Both successful and failed actions contribute to total duration
|
||||
assert_eq!(memory.total_duration(), Duration::from_millis(300));
|
||||
assert_eq!(memory.total_duration(), Duration::from_millis(300)); // safety: test
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,10 +6,12 @@
|
||||
//! - State machine
|
||||
//! - Resource tracking
|
||||
|
||||
pub mod fallback;
|
||||
mod manager;
|
||||
mod memory;
|
||||
mod state;
|
||||
|
||||
pub use fallback::FallbackDeliverable;
|
||||
pub use manager::ContextManager;
|
||||
pub use memory::{ActionRecord, ConversationMemory, Memory};
|
||||
pub use state::{JobContext, JobState, StateTransition, TokenBudgetExceeded};
|
||||
|
||||
@@ -333,6 +333,12 @@ async fn job_event_handler(
|
||||
.get("session_id")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|s| s.to_string()),
|
||||
// NOTE: `fallback_deliverable` is currently always None in SSE events.
|
||||
// In-memory jobs store fallback data in JobContext.metadata (accessed via job_status tool).
|
||||
// Sandbox containers don't yet emit fallback data in their event payloads.
|
||||
// This field is forward-compatible infrastructure for when container workers
|
||||
// gain context/memory tracking capabilities.
|
||||
fallback_deliverable: payload.data.get("fallback_deliverable").cloned(),
|
||||
},
|
||||
_ => SseEvent::JobStatus {
|
||||
job_id: job_id_str,
|
||||
|
||||
+181
-119
@@ -1005,7 +1005,8 @@ impl Tool for JobStatusTool {
|
||||
"created_at": job_ctx.created_at.to_rfc3339(),
|
||||
"started_at": job_ctx.started_at.map(|t| t.to_rfc3339()),
|
||||
"completed_at": job_ctx.completed_at.map(|t| t.to_rfc3339()),
|
||||
"actual_cost": job_ctx.actual_cost.to_string()
|
||||
"actual_cost": job_ctx.actual_cost.to_string(),
|
||||
"fallback_deliverable": job_ctx.metadata.get("fallback_deliverable"),
|
||||
});
|
||||
Ok(ToolOutput::success(result, start.elapsed()))
|
||||
}
|
||||
@@ -1384,7 +1385,7 @@ mod tests {
|
||||
let tool = CreateJobTool::new(manager.clone());
|
||||
|
||||
// Without sandbox deps, it should use the local path
|
||||
assert!(!tool.sandbox_enabled());
|
||||
assert!(!tool.sandbox_enabled()); // safety: test
|
||||
|
||||
let params = serde_json::json!({
|
||||
"title": "Test Job",
|
||||
@@ -1392,12 +1393,13 @@ mod tests {
|
||||
});
|
||||
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(params, &ctx).await.unwrap();
|
||||
let result = tool.execute(params, &ctx).await.unwrap(); // safety: test
|
||||
|
||||
let job_id = result.result.get("job_id").unwrap().as_str().unwrap();
|
||||
assert!(!job_id.is_empty());
|
||||
let job_id = result.result.get("job_id").unwrap().as_str().unwrap(); // safety: test
|
||||
assert!(!job_id.is_empty()); // safety: test
|
||||
assert_eq!(
|
||||
result.result.get("status").unwrap().as_str().unwrap(),
|
||||
/* safety: test */
|
||||
result.result.get("status").unwrap().as_str().unwrap(), // safety: test
|
||||
"pending"
|
||||
);
|
||||
}
|
||||
@@ -1409,11 +1411,11 @@ mod tests {
|
||||
// Without sandbox
|
||||
let tool = CreateJobTool::new(Arc::clone(&manager));
|
||||
let schema = tool.parameters_schema();
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap();
|
||||
assert!(props.contains_key("title"));
|
||||
assert!(props.contains_key("description"));
|
||||
assert!(!props.contains_key("wait"));
|
||||
assert!(!props.contains_key("mode"));
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap(); // safety: test
|
||||
assert!(props.contains_key("title")); // safety: test
|
||||
assert!(props.contains_key("description")); // safety: test
|
||||
assert!(!props.contains_key("wait")); // safety: test
|
||||
assert!(!props.contains_key("mode")); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -1422,7 +1424,7 @@ mod tests {
|
||||
|
||||
// Without sandbox: default timeout
|
||||
let tool = CreateJobTool::new(Arc::clone(&manager));
|
||||
assert_eq!(tool.execution_timeout(), Duration::from_secs(30));
|
||||
assert_eq!(tool.execution_timeout(), Duration::from_secs(30)); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1455,23 +1457,23 @@ mod tests {
|
||||
let manager = Arc::new(ContextManager::new(5));
|
||||
|
||||
// Create some jobs
|
||||
manager.create_job("Job 1", "Desc 1").await.unwrap();
|
||||
manager.create_job("Job 2", "Desc 2").await.unwrap();
|
||||
manager.create_job("Job 1", "Desc 1").await.unwrap(); // safety: test
|
||||
manager.create_job("Job 2", "Desc 2").await.unwrap(); // safety: test
|
||||
|
||||
let tool = ListJobsTool::new(manager);
|
||||
|
||||
let params = serde_json::json!({});
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(params, &ctx).await.unwrap();
|
||||
let result = tool.execute(params, &ctx).await.unwrap(); // safety: test
|
||||
|
||||
let jobs = result.result.get("jobs").unwrap().as_array().unwrap();
|
||||
assert_eq!(jobs.len(), 2);
|
||||
let jobs = result.result.get("jobs").unwrap().as_array().unwrap(); // safety: test
|
||||
assert_eq!(jobs.len(), 2); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_job_status_tool() {
|
||||
let manager = Arc::new(ContextManager::new(5));
|
||||
let job_id = manager.create_job("Test Job", "Description").await.unwrap();
|
||||
let job_id = manager.create_job("Test Job", "Description").await.unwrap(); // safety: test
|
||||
|
||||
let tool = JobStatusTool::new(manager);
|
||||
|
||||
@@ -1479,10 +1481,11 @@ mod tests {
|
||||
"job_id": job_id.to_string()
|
||||
});
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(params, &ctx).await.unwrap();
|
||||
let result = tool.execute(params, &ctx).await.unwrap(); // safety: test
|
||||
|
||||
assert_eq!(
|
||||
result.result.get("title").unwrap().as_str().unwrap(),
|
||||
/* safety: test */
|
||||
result.result.get("title").unwrap().as_str().unwrap(), // safety: test
|
||||
"Test Job"
|
||||
);
|
||||
}
|
||||
@@ -1496,8 +1499,9 @@ mod tests {
|
||||
let missing_title = tool
|
||||
.execute(serde_json::json!({ "description": "A test job" }), &ctx)
|
||||
.await;
|
||||
assert!(missing_title.is_err());
|
||||
assert!(missing_title.is_err()); // safety: test
|
||||
assert!(
|
||||
/* safety: test */
|
||||
missing_title
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
@@ -1507,8 +1511,9 @@ mod tests {
|
||||
let missing_description = tool
|
||||
.execute(serde_json::json!({ "title": "Test Job" }), &ctx)
|
||||
.await;
|
||||
assert!(missing_description.is_err());
|
||||
assert!(missing_description.is_err()); // safety: test
|
||||
assert!(
|
||||
/* safety: test */
|
||||
missing_description
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
@@ -1522,19 +1527,19 @@ mod tests {
|
||||
let pending_id = manager
|
||||
.create_job_for_user("default", "Pending Job", "Todo")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
let completed_id = manager
|
||||
.create_job_for_user("default", "Completed Job", "Done")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
let failed_id = manager
|
||||
.create_job_for_user("default", "Failed Job", "Oops")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
manager
|
||||
.create_job_for_user("other-user", "Other User Job", "Ignore")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
manager
|
||||
.update_context(completed_id, |ctx| {
|
||||
@@ -1542,41 +1547,44 @@ mod tests {
|
||||
ctx.transition_to(JobState::Completed, Some("done".to_string()))
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
manager
|
||||
.update_context(failed_id, |ctx| {
|
||||
ctx.transition_to(JobState::InProgress, None)?;
|
||||
ctx.transition_to(JobState::Failed, Some("boom".to_string()))
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let tool = ListJobsTool::new(Arc::clone(&manager));
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(serde_json::json!({}), &ctx).await.unwrap();
|
||||
let result = tool.execute(serde_json::json!({}), &ctx).await.unwrap(); // safety: test
|
||||
|
||||
let jobs = result.result.get("jobs").unwrap().as_array().unwrap();
|
||||
assert_eq!(jobs.len(), 3);
|
||||
let jobs = result.result.get("jobs").unwrap().as_array().unwrap(); // safety: test
|
||||
assert_eq!(jobs.len(), 3); // safety: test
|
||||
assert!(jobs.iter().any(|job| {
|
||||
// safety: test
|
||||
job.get("job_id").and_then(|v| v.as_str()) == Some(&pending_id.to_string())
|
||||
&& job.get("status").and_then(|v| v.as_str()) == Some("Pending")
|
||||
}));
|
||||
assert!(jobs.iter().any(|job| {
|
||||
// safety: test
|
||||
job.get("job_id").and_then(|v| v.as_str()) == Some(&completed_id.to_string())
|
||||
&& job.get("status").and_then(|v| v.as_str()) == Some("Completed")
|
||||
}));
|
||||
assert!(jobs.iter().any(|job| {
|
||||
// safety: test
|
||||
job.get("job_id").and_then(|v| v.as_str()) == Some(&failed_id.to_string())
|
||||
&& job.get("status").and_then(|v| v.as_str()) == Some("Failed")
|
||||
}));
|
||||
|
||||
let summary = result.result.get("summary").unwrap();
|
||||
assert_eq!(summary.get("total").and_then(|v| v.as_u64()), Some(3));
|
||||
assert_eq!(summary.get("pending").and_then(|v| v.as_u64()), Some(1));
|
||||
assert_eq!(summary.get("completed").and_then(|v| v.as_u64()), Some(1));
|
||||
assert_eq!(summary.get("failed").and_then(|v| v.as_u64()), Some(1));
|
||||
let summary = result.result.get("summary").unwrap(); // safety: test
|
||||
assert_eq!(summary.get("total").and_then(|v| v.as_u64()), Some(3)); // safety: test
|
||||
assert_eq!(summary.get("pending").and_then(|v| v.as_u64()), Some(1)); // safety: test
|
||||
assert_eq!(summary.get("completed").and_then(|v| v.as_u64()), Some(1)); // safety: test
|
||||
assert_eq!(summary.get("failed").and_then(|v| v.as_u64()), Some(1)); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1585,29 +1593,30 @@ mod tests {
|
||||
let job_id = manager
|
||||
.create_job_for_user("default", "Transition Job", "Track me")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
manager
|
||||
.update_context(job_id, |ctx| {
|
||||
ctx.transition_to(JobState::InProgress, Some("started".to_string()))?;
|
||||
ctx.transition_to(JobState::Completed, Some("finished".to_string()))
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let tool = JobStatusTool::new(Arc::clone(&manager));
|
||||
let ctx = JobContext::default();
|
||||
let result = tool
|
||||
.execute(serde_json::json!({ "job_id": job_id.to_string() }), &ctx)
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
assert_eq!(
|
||||
/* safety: test */
|
||||
result.result.get("status").and_then(|v| v.as_str()),
|
||||
Some("Completed")
|
||||
);
|
||||
assert!(result.result.get("started_at").unwrap().is_string());
|
||||
assert!(result.result.get("completed_at").unwrap().is_string());
|
||||
assert!(result.result.get("started_at").unwrap().is_string()); // safety: test
|
||||
assert!(result.result.get("completed_at").unwrap().is_string()); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1616,26 +1625,27 @@ mod tests {
|
||||
let job_id = manager
|
||||
.create_job_for_user("default", "Running Job", "In progress")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
manager
|
||||
.update_context(job_id, |ctx| ctx.transition_to(JobState::InProgress, None))
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let tool = CancelJobTool::new(Arc::clone(&manager));
|
||||
let ctx = JobContext::default();
|
||||
let result = tool
|
||||
.execute(serde_json::json!({ "job_id": job_id.to_string() }), &ctx)
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
assert_eq!(
|
||||
/* safety: test */
|
||||
result.result.get("status").and_then(|v| v.as_str()),
|
||||
Some("cancelled")
|
||||
);
|
||||
let updated = manager.get_context(job_id).await.unwrap();
|
||||
assert_eq!(updated.state, JobState::Cancelled);
|
||||
let updated = manager.get_context(job_id).await.unwrap(); // safety: test
|
||||
assert_eq!(updated.state, JobState::Cancelled); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1644,39 +1654,81 @@ mod tests {
|
||||
let job_id = manager
|
||||
.create_job_for_user("default", "Completed Job", "Already done")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
manager
|
||||
.update_context(job_id, |ctx| {
|
||||
ctx.transition_to(JobState::InProgress, None)?;
|
||||
ctx.transition_to(JobState::Completed, Some("done".to_string()))
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let tool = CancelJobTool::new(Arc::clone(&manager));
|
||||
let ctx = JobContext::default();
|
||||
let result = tool
|
||||
.execute(serde_json::json!({ "job_id": job_id.to_string() }), &ctx)
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let error = result.result.get("error").and_then(|v| v.as_str()).unwrap();
|
||||
assert!(error.contains("Cannot cancel job"));
|
||||
assert!(error.contains("completed"));
|
||||
let error = result.result.get("error").and_then(|v| v.as_str()).unwrap(); // safety: test
|
||||
assert!(error.contains("Cannot cancel job")); // safety: test
|
||||
assert!(error.contains("completed")); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_job_status_includes_fallback_deliverable() {
|
||||
let manager = Arc::new(ContextManager::new(5));
|
||||
let job_id = manager
|
||||
.create_job_for_user("default", "Failing Job", "Will fail")
|
||||
.await
|
||||
.unwrap(); // safety: test
|
||||
|
||||
// Inject a real FallbackDeliverable into the job metadata.
|
||||
let fallback = serde_json::json!({
|
||||
"partial": true,
|
||||
"failure_reason": "max iterations",
|
||||
"last_action": null,
|
||||
"action_stats": { "total": 5, "successful": 3, "failed": 2 },
|
||||
"tokens_used": 1000,
|
||||
"cost": "0.05",
|
||||
"elapsed_secs": 12.5,
|
||||
"repair_attempts": 1,
|
||||
});
|
||||
manager
|
||||
.update_context(job_id, |ctx| {
|
||||
ctx.metadata = serde_json::json!({ "fallback_deliverable": fallback.clone() });
|
||||
Ok::<(), String>(())
|
||||
})
|
||||
.await
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let tool = JobStatusTool::new(manager);
|
||||
let params = serde_json::json!({ "job_id": job_id.to_string() });
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(params, &ctx).await.unwrap(); // safety: test
|
||||
|
||||
let fb = result.result.get("fallback_deliverable").unwrap(); // safety: test
|
||||
assert_eq!(fb.get("partial").unwrap(), true); // safety: test
|
||||
assert_eq!(fb.get("failure_reason").unwrap(), "max iterations"); // safety: test
|
||||
let stats = fb.get("action_stats").unwrap(); // safety: test
|
||||
assert_eq!(stats.get("total").unwrap(), 5); // safety: test
|
||||
assert_eq!(stats.get("successful").unwrap(), 3); // safety: test
|
||||
assert_eq!(stats.get("failed").unwrap(), 2); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_project_dir_auto() {
|
||||
let project_id = Uuid::new_v4();
|
||||
let (dir, browse_id) = resolve_project_dir(None, project_id).unwrap();
|
||||
assert!(dir.exists());
|
||||
assert!(dir.ends_with(project_id.to_string()));
|
||||
assert_eq!(browse_id, project_id.to_string());
|
||||
let (dir, browse_id) = resolve_project_dir(None, project_id).unwrap(); // safety: test
|
||||
assert!(dir.exists()); // safety: test
|
||||
assert!(dir.ends_with(project_id.to_string())); // safety: test
|
||||
assert_eq!(browse_id, project_id.to_string()); // safety: test
|
||||
|
||||
// Must be under the projects base
|
||||
let base = projects_base().canonicalize().unwrap();
|
||||
assert!(dir.starts_with(&base));
|
||||
let base = projects_base().canonicalize().unwrap(); // safety: test
|
||||
assert!(dir.starts_with(&base)); // safety: test
|
||||
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
}
|
||||
@@ -1684,33 +1736,34 @@ mod tests {
|
||||
#[test]
|
||||
fn test_resolve_project_dir_explicit_under_base() {
|
||||
let base = projects_base();
|
||||
std::fs::create_dir_all(&base).unwrap();
|
||||
std::fs::create_dir_all(&base).unwrap(); // safety: test
|
||||
let explicit = base.join("test_explicit_project");
|
||||
// Explicit paths must already exist (no auto-create).
|
||||
std::fs::create_dir_all(&explicit).unwrap();
|
||||
std::fs::create_dir_all(&explicit).unwrap(); // safety: test
|
||||
let project_id = Uuid::new_v4();
|
||||
|
||||
let (dir, browse_id) = resolve_project_dir(Some(explicit.clone()), project_id).unwrap();
|
||||
assert!(dir.exists());
|
||||
assert_eq!(browse_id, "test_explicit_project");
|
||||
let (dir, browse_id) = resolve_project_dir(Some(explicit.clone()), project_id).unwrap(); // safety: test
|
||||
assert!(dir.exists()); // safety: test
|
||||
assert_eq!(browse_id, "test_explicit_project"); // safety: test
|
||||
|
||||
let canonical_base = base.canonicalize().unwrap();
|
||||
assert!(dir.starts_with(&canonical_base));
|
||||
let canonical_base = base.canonicalize().unwrap(); // safety: test
|
||||
assert!(dir.starts_with(&canonical_base)); // safety: test
|
||||
|
||||
let _ = std::fs::remove_dir_all(&explicit);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resolve_project_dir_rejects_outside_base() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let tmp = tempfile::tempdir().unwrap(); // safety: test
|
||||
let escape_attempt = tmp.path().join("evil_project");
|
||||
// Don't create it: explicit paths that don't exist are rejected
|
||||
// before the prefix check even runs.
|
||||
|
||||
let result = resolve_project_dir(Some(escape_attempt), Uuid::new_v4());
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
/* safety: test */
|
||||
err.contains("does not exist"),
|
||||
"expected 'does not exist' error, got: {}",
|
||||
err
|
||||
@@ -1720,13 +1773,14 @@ mod tests {
|
||||
#[test]
|
||||
fn test_resolve_project_dir_rejects_outside_base_existing() {
|
||||
// A directory that exists but is outside the projects base.
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let tmp = tempfile::tempdir().unwrap(); // safety: test
|
||||
let outside = tmp.path().to_path_buf();
|
||||
|
||||
let result = resolve_project_dir(Some(outside), Uuid::new_v4());
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
/* safety: test */
|
||||
err.contains("must be under"),
|
||||
"expected 'must be under' error, got: {}",
|
||||
err
|
||||
@@ -1740,7 +1794,7 @@ mod tests {
|
||||
let traversal = base.join("legit").join("..").join("..").join(".ssh");
|
||||
|
||||
let result = resolve_project_dir(Some(traversal), Uuid::new_v4());
|
||||
assert!(result.is_err(), "traversal path should be rejected");
|
||||
assert!(result.is_err(), "traversal path should be rejected"); // safety: test
|
||||
|
||||
// Traversal path that actually resolves gets the prefix check.
|
||||
// `base/../` resolves to the parent of projects base, which is outside.
|
||||
@@ -1748,7 +1802,7 @@ mod tests {
|
||||
std::fs::create_dir_all(&base_parent).ok();
|
||||
if base_parent.exists() {
|
||||
let result = resolve_project_dir(Some(base_parent.clone()), Uuid::new_v4());
|
||||
assert!(result.is_err(), "path outside base should be rejected");
|
||||
assert!(result.is_err(), "path outside base should be rejected"); // safety: test
|
||||
let _ = std::fs::remove_dir_all(&base_parent);
|
||||
}
|
||||
}
|
||||
@@ -1762,8 +1816,9 @@ mod tests {
|
||||
));
|
||||
let tool = CreateJobTool::new(manager).with_sandbox(jm, None);
|
||||
let schema = tool.parameters_schema();
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap();
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap(); // safety: test
|
||||
assert!(
|
||||
/* safety: test */
|
||||
props.contains_key("project_dir"),
|
||||
"sandbox schema must expose project_dir"
|
||||
);
|
||||
@@ -1778,8 +1833,9 @@ mod tests {
|
||||
));
|
||||
let tool = CreateJobTool::new(manager).with_sandbox(jm, None);
|
||||
let schema = tool.parameters_schema();
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap();
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap(); // safety: test
|
||||
assert!(
|
||||
/* safety: test */
|
||||
props.contains_key("credentials"),
|
||||
"sandbox schema must expose credentials"
|
||||
);
|
||||
@@ -1792,13 +1848,13 @@ mod tests {
|
||||
|
||||
// No credentials parameter
|
||||
let params = serde_json::json!({"title": "t", "description": "d"});
|
||||
let grants = tool.parse_credentials(¶ms, "user1").await.unwrap();
|
||||
assert!(grants.is_empty());
|
||||
let grants = tool.parse_credentials(¶ms, "user1").await.unwrap(); // safety: test
|
||||
assert!(grants.is_empty()); // safety: test
|
||||
|
||||
// Empty credentials object
|
||||
let params = serde_json::json!({"credentials": {}});
|
||||
let grants = tool.parse_credentials(¶ms, "user1").await.unwrap();
|
||||
assert!(grants.is_empty());
|
||||
let grants = tool.parse_credentials(¶ms, "user1").await.unwrap(); // safety: test
|
||||
assert!(grants.is_empty()); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1808,9 +1864,10 @@ mod tests {
|
||||
|
||||
let params = serde_json::json!({"credentials": {"my_secret": "MY_SECRET"}});
|
||||
let result = tool.parse_credentials(¶ms, "user1").await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
/* safety: test */
|
||||
err.contains("no secrets store"),
|
||||
"expected 'no secrets store' error, got: {}",
|
||||
err
|
||||
@@ -1828,9 +1885,10 @@ mod tests {
|
||||
|
||||
let params = serde_json::json!({"credentials": {"nonexistent_secret": "SOME_VAR"}});
|
||||
let result = tool.parse_credentials(¶ms, "user1").await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
/* safety: test */
|
||||
err.contains("not found"),
|
||||
"expected 'not found' error, got: {}",
|
||||
err
|
||||
@@ -1852,17 +1910,17 @@ mod tests {
|
||||
CreateSecretParams::new("github_token", TEST_GITHUB_TOKEN),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let tool = CreateJobTool::new(manager).with_secrets(Arc::clone(&secrets));
|
||||
|
||||
let params = serde_json::json!({
|
||||
"credentials": {"github_token": "GITHUB_TOKEN"}
|
||||
});
|
||||
let grants = tool.parse_credentials(¶ms, "user1").await.unwrap();
|
||||
assert_eq!(grants.len(), 1);
|
||||
assert_eq!(grants[0].secret_name, "github_token");
|
||||
assert_eq!(grants[0].env_var, "GITHUB_TOKEN");
|
||||
let grants = tool.parse_credentials(¶ms, "user1").await.unwrap(); // safety: test
|
||||
assert_eq!(grants.len(), 1); // safety: test
|
||||
assert_eq!(grants[0].secret_name, "github_token"); // safety: test
|
||||
assert_eq!(grants[0].env_var, "GITHUB_TOKEN"); // safety: test
|
||||
}
|
||||
|
||||
fn test_prompt_tool(queue: PromptQueue) -> JobPromptTool {
|
||||
@@ -1876,7 +1934,7 @@ mod tests {
|
||||
let job_id = cm
|
||||
.create_job_for_user("default", "Test Job", "desc")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let queue: PromptQueue =
|
||||
Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));
|
||||
@@ -1889,18 +1947,19 @@ mod tests {
|
||||
});
|
||||
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(params, &ctx).await.unwrap();
|
||||
let result = tool.execute(params, &ctx).await.unwrap(); // safety: test
|
||||
|
||||
assert_eq!(
|
||||
result.result.get("status").unwrap().as_str().unwrap(),
|
||||
/* safety: test */
|
||||
result.result.get("status").unwrap().as_str().unwrap(), // safety: test
|
||||
"queued"
|
||||
);
|
||||
|
||||
let q = queue.lock().await;
|
||||
let prompts = q.get(&job_id).unwrap();
|
||||
assert_eq!(prompts.len(), 1);
|
||||
assert_eq!(prompts[0].content, "What's the status?");
|
||||
assert!(!prompts[0].done);
|
||||
let prompts = q.get(&job_id).unwrap(); // safety: test
|
||||
assert_eq!(prompts.len(), 1); // safety: test
|
||||
assert_eq!(prompts[0].content, "What's the status?"); // safety: test
|
||||
assert!(!prompts[0].done); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1910,6 +1969,7 @@ mod tests {
|
||||
Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));
|
||||
let tool = test_prompt_tool(queue);
|
||||
assert_eq!(
|
||||
/* safety: test */
|
||||
tool.requires_approval(&serde_json::json!({})),
|
||||
ApprovalRequirement::UnlessAutoApproved
|
||||
);
|
||||
@@ -1928,7 +1988,7 @@ mod tests {
|
||||
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(params, &ctx).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1943,7 +2003,7 @@ mod tests {
|
||||
|
||||
let ctx = JobContext::default();
|
||||
let result = tool.execute(params, &ctx).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1958,7 +2018,7 @@ mod tests {
|
||||
let job_id = cm
|
||||
.create_job_for_user("owner-user", "Secret Job", "classified")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
// We need a Store to construct the tool, but creating one requires
|
||||
// a database URL. Instead, test the ownership logic directly:
|
||||
@@ -1968,9 +2028,9 @@ mod tests {
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let job_ctx = cm.get_context(job_id).await.unwrap();
|
||||
assert_ne!(job_ctx.user_id, attacker_ctx.user_id);
|
||||
assert_eq!(job_ctx.user_id, "owner-user");
|
||||
let job_ctx = cm.get_context(job_id).await.unwrap(); // safety: test
|
||||
assert_ne!(job_ctx.user_id, attacker_ctx.user_id); // safety: test
|
||||
assert_eq!(job_ctx.user_id, "owner-user"); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -1991,12 +2051,12 @@ mod tests {
|
||||
"required": ["job_id"]
|
||||
});
|
||||
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap();
|
||||
assert!(props.contains_key("job_id"));
|
||||
assert!(props.contains_key("limit"));
|
||||
let required = schema.get("required").unwrap().as_array().unwrap();
|
||||
assert_eq!(required.len(), 1);
|
||||
assert_eq!(required[0].as_str().unwrap(), "job_id");
|
||||
let props = schema.get("properties").unwrap().as_object().unwrap(); // safety: test
|
||||
assert!(props.contains_key("job_id")); // safety: test
|
||||
assert!(props.contains_key("limit")); // safety: test
|
||||
let required = schema.get("required").unwrap().as_array().unwrap(); // safety: test
|
||||
assert_eq!(required.len(), 1); // safety: test
|
||||
assert_eq!(required[0].as_str().unwrap(), "job_id"); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -2005,7 +2065,7 @@ mod tests {
|
||||
let job_id = cm
|
||||
.create_job_for_user("owner-user", "Test Job", "desc")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let queue: PromptQueue =
|
||||
Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));
|
||||
@@ -2023,9 +2083,10 @@ mod tests {
|
||||
};
|
||||
|
||||
let result = tool.execute(params, &ctx).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
/* safety: test */
|
||||
err.contains("does not belong to current user"),
|
||||
"expected ownership error, got: {}",
|
||||
err
|
||||
@@ -2035,33 +2096,34 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn test_resolve_job_id_full_uuid() {
|
||||
let cm = ContextManager::new(5);
|
||||
let job_id = cm.create_job("Test", "Desc").await.unwrap();
|
||||
let job_id = cm.create_job("Test", "Desc").await.unwrap(); // safety: test
|
||||
|
||||
let resolved = resolve_job_id(&job_id.to_string(), &cm).await.unwrap();
|
||||
assert_eq!(resolved, job_id);
|
||||
let resolved = resolve_job_id(&job_id.to_string(), &cm).await.unwrap(); // safety: test
|
||||
assert_eq!(resolved, job_id); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_job_id_short_prefix() {
|
||||
let cm = ContextManager::new(5);
|
||||
let job_id = cm.create_job("Test", "Desc").await.unwrap();
|
||||
let job_id = cm.create_job("Test", "Desc").await.unwrap(); // safety: test
|
||||
|
||||
// Use first 8 hex chars (without dashes)
|
||||
let hex = job_id.to_string().replace('-', "");
|
||||
let prefix = &hex[..8];
|
||||
let resolved = resolve_job_id(prefix, &cm).await.unwrap();
|
||||
assert_eq!(resolved, job_id);
|
||||
let resolved = resolve_job_id(prefix, &cm).await.unwrap(); // safety: test
|
||||
assert_eq!(resolved, job_id); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resolve_job_id_no_match() {
|
||||
let cm = ContextManager::new(5);
|
||||
cm.create_job("Test", "Desc").await.unwrap();
|
||||
cm.create_job("Test", "Desc").await.unwrap(); // safety: test
|
||||
|
||||
let result = resolve_job_id("00000000", &cm).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
/* safety: test */
|
||||
err.contains("no job found"),
|
||||
"expected 'no job found', got: {}",
|
||||
err
|
||||
@@ -2072,6 +2134,6 @@ mod tests {
|
||||
async fn test_resolve_job_id_invalid_input() {
|
||||
let cm = ContextManager::new(5);
|
||||
let result = resolve_job_id("not-hex-at-all!", &cm).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.is_err()); // safety: test
|
||||
}
|
||||
}
|
||||
|
||||
+149
-28
@@ -196,6 +196,7 @@ impl Worker {
|
||||
.get("session_id")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|s| s.to_string()),
|
||||
fallback_deliverable: data.get("fallback_deliverable").cloned(),
|
||||
}),
|
||||
_ => None,
|
||||
};
|
||||
@@ -960,9 +961,14 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
}
|
||||
|
||||
async fn mark_failed(&self, reason: &str) -> Result<(), Error> {
|
||||
// Build fallback deliverable from memory before transitioning.
|
||||
let fallback = self.build_fallback(reason).await;
|
||||
|
||||
self.context_manager()
|
||||
.update_context(self.job_id, |ctx| {
|
||||
ctx.transition_to(JobState::Failed, Some(reason.to_string()))
|
||||
ctx.transition_to(JobState::Failed, Some(reason.to_string()))?;
|
||||
store_fallback_in_metadata(ctx, fallback.as_ref());
|
||||
Ok(())
|
||||
})
|
||||
.await?
|
||||
.map_err(|s| crate::error::JobError::ContextError {
|
||||
@@ -983,8 +989,15 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
}
|
||||
|
||||
async fn mark_stuck(&self, reason: &str) -> Result<(), Error> {
|
||||
// Build fallback deliverable from memory before transitioning.
|
||||
let fallback = self.build_fallback(reason).await;
|
||||
|
||||
self.context_manager()
|
||||
.update_context(self.job_id, |ctx| ctx.mark_stuck(reason))
|
||||
.update_context(self.job_id, |ctx| {
|
||||
ctx.mark_stuck(reason)?;
|
||||
store_fallback_in_metadata(ctx, fallback.as_ref());
|
||||
Ok(())
|
||||
})
|
||||
.await?
|
||||
.map_err(|s| crate::error::JobError::ContextError {
|
||||
id: self.job_id,
|
||||
@@ -1002,6 +1015,57 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
self.persist_status(JobState::Stuck, Some(reason.to_string()));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Build a [`FallbackDeliverable`] from the current job context and memory.
|
||||
async fn build_fallback(&self, reason: &str) -> Option<crate::context::FallbackDeliverable> {
|
||||
let memory = match self.context_manager().get_memory(self.job_id).await {
|
||||
Ok(memory) => memory,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
job_id = %self.job_id,
|
||||
"Failed to load memory while building fallback deliverable: {e}"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
let ctx = match self.context_manager().get_context(self.job_id).await {
|
||||
Ok(ctx) => ctx,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
job_id = %self.job_id,
|
||||
"Failed to load context while building fallback deliverable: {e}"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
Some(crate::context::FallbackDeliverable::build(
|
||||
&ctx, &memory, reason,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
/// Store a fallback deliverable in the job context's metadata.
|
||||
fn store_fallback_in_metadata(
|
||||
ctx: &mut crate::context::JobContext,
|
||||
fallback: Option<&crate::context::FallbackDeliverable>,
|
||||
) {
|
||||
let Some(fb) = fallback else {
|
||||
return;
|
||||
};
|
||||
match serde_json::to_value(fb) {
|
||||
Ok(val) => {
|
||||
if !ctx.metadata.is_object() {
|
||||
ctx.metadata = serde_json::json!({});
|
||||
}
|
||||
ctx.metadata["fallback_deliverable"] = val;
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"Failed to serialize fallback deliverable for job {}: {e}",
|
||||
ctx.job_id
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Job delegate: implements `LoopDelegate` for the background job context.
|
||||
@@ -1440,7 +1504,7 @@ mod tests {
|
||||
}
|
||||
|
||||
let cm = Arc::new(crate::context::ContextManager::new(5));
|
||||
let job_id = cm.create_job("test", "test job").await.unwrap();
|
||||
let job_id = cm.create_job("test", "test job").await.unwrap(); // safety: test
|
||||
|
||||
let deps = WorkerDeps {
|
||||
context_manager: cm,
|
||||
@@ -1472,8 +1536,9 @@ mod tests {
|
||||
tool_call_id: "call_abc123".to_string(),
|
||||
};
|
||||
|
||||
assert_eq!(selection.tool_call_id, "call_abc123");
|
||||
assert_eq!(selection.tool_call_id, "call_abc123"); // safety: test
|
||||
assert_ne!(
|
||||
/* safety: test */
|
||||
selection.tool_call_id, "tool_call_id",
|
||||
"tool_call_id must not be the hardcoded placeholder string"
|
||||
);
|
||||
@@ -1509,11 +1574,12 @@ mod tests {
|
||||
let results = worker.execute_tools_parallel(&selections).await;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
assert_eq!(results.len(), 3);
|
||||
assert_eq!(results.len(), 3); // safety: test
|
||||
for r in &results {
|
||||
assert!(r.result.is_ok(), "Tool should succeed");
|
||||
assert!(r.result.is_ok(), "Tool should succeed"); // safety: test
|
||||
}
|
||||
assert!(
|
||||
/* safety: test */
|
||||
elapsed < Duration::from_millis(800),
|
||||
"Parallel execution took {:?}, expected < 800ms (sequential would be ~600ms)",
|
||||
elapsed
|
||||
@@ -1565,9 +1631,9 @@ mod tests {
|
||||
|
||||
let results = worker.execute_tools_parallel(&selections).await;
|
||||
|
||||
assert!(results[0].result.as_ref().unwrap().contains("done_tool_a"));
|
||||
assert!(results[1].result.as_ref().unwrap().contains("done_tool_b"));
|
||||
assert!(results[2].result.as_ref().unwrap().contains("done_tool_c"));
|
||||
assert!(results[0].result.as_ref().unwrap().contains("done_tool_a")); // safety: test
|
||||
assert!(results[1].result.as_ref().unwrap().contains("done_tool_b")); // safety: test
|
||||
assert!(results[2].result.as_ref().unwrap().contains("done_tool_c")); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1583,8 +1649,9 @@ mod tests {
|
||||
}];
|
||||
|
||||
let results = worker.execute_tools_parallel(&selections).await;
|
||||
assert_eq!(results.len(), 1);
|
||||
assert_eq!(results.len(), 1); // safety: test
|
||||
assert!(
|
||||
/* safety: test */
|
||||
results[0].result.is_err(),
|
||||
"Missing tool should produce an error, not a panic"
|
||||
);
|
||||
@@ -1600,23 +1667,24 @@ mod tests {
|
||||
ctx.transition_to(JobState::InProgress, None)
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
worker.mark_completed().await.unwrap();
|
||||
worker.mark_completed().await.unwrap(); // safety: test
|
||||
|
||||
let ctx = worker
|
||||
.context_manager()
|
||||
.get_context(worker.job_id)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(ctx.state, JobState::Completed);
|
||||
.unwrap(); // safety: test
|
||||
assert_eq!(ctx.state, JobState::Completed); // safety: test
|
||||
|
||||
// Second mark_completed should succeed (idempotent) rather than
|
||||
// erroring, matching the fix for the execution_loop / worker wrapper
|
||||
// race condition.
|
||||
let result = worker.mark_completed().await;
|
||||
assert!(
|
||||
/* safety: test */
|
||||
result.is_ok(),
|
||||
"Completed -> Completed transition should be idempotent"
|
||||
);
|
||||
@@ -1641,7 +1709,7 @@ mod tests {
|
||||
}
|
||||
|
||||
let cm = Arc::new(crate::context::ContextManager::new(5));
|
||||
let job_id = cm.create_job("test", "test job").await.unwrap();
|
||||
let job_id = cm.create_job("test", "test job").await.unwrap(); // safety: test
|
||||
|
||||
let deps = WorkerDeps {
|
||||
context_manager: cm,
|
||||
@@ -1740,6 +1808,7 @@ mod tests {
|
||||
.execute_tool("needs_approval", &serde_json::json!({}))
|
||||
.await;
|
||||
assert!(
|
||||
/* safety: test */
|
||||
result.is_err(),
|
||||
"Should be blocked without approval context"
|
||||
);
|
||||
@@ -1752,7 +1821,7 @@ mod tests {
|
||||
let result = worker_allowed
|
||||
.execute_tool("needs_approval", &serde_json::json!({}))
|
||||
.await;
|
||||
assert!(result.is_ok(), "Should be allowed with autonomous context");
|
||||
assert!(result.is_ok(), "Should be allowed with autonomous context"); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1766,6 +1835,7 @@ mod tests {
|
||||
.execute_tool("always_approval", &serde_json::json!({}))
|
||||
.await;
|
||||
assert!(
|
||||
/* safety: test */
|
||||
result.is_err(),
|
||||
"Always tool should be blocked without permission"
|
||||
);
|
||||
@@ -1781,6 +1851,7 @@ mod tests {
|
||||
.execute_tool("always_approval", &serde_json::json!({}))
|
||||
.await;
|
||||
assert!(
|
||||
/* safety: test */
|
||||
result.is_ok(),
|
||||
"Always tool should be allowed with permission"
|
||||
);
|
||||
@@ -1797,8 +1868,8 @@ mod tests {
|
||||
ctx.transition_to(JobState::InProgress, None)
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
// Set a token budget
|
||||
worker
|
||||
@@ -1807,16 +1878,17 @@ mod tests {
|
||||
ctx.max_tokens = 100;
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
// Simulate adding tokens that exceed the budget
|
||||
let budget_result = worker
|
||||
.context_manager()
|
||||
.update_context(worker.job_id, |ctx| ctx.add_tokens(200))
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
assert!(
|
||||
/* safety: test */
|
||||
budget_result.is_err(),
|
||||
"Should return error when token budget exceeded"
|
||||
);
|
||||
@@ -1825,13 +1897,13 @@ mod tests {
|
||||
worker
|
||||
.mark_failed(&budget_result.unwrap_err().to_string())
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
let ctx = worker
|
||||
.context_manager()
|
||||
.get_context(worker.job_id)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(ctx.state, JobState::Failed);
|
||||
.unwrap(); // safety: test
|
||||
assert_eq!(ctx.state, JobState::Failed); // safety: test
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1845,21 +1917,22 @@ mod tests {
|
||||
ctx.transition_to(JobState::InProgress, None)
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
// Simulate what the execution loop does when max_iterations is exceeded
|
||||
worker
|
||||
.mark_failed("Maximum iterations exceeded: job hit the iteration cap")
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let ctx = worker
|
||||
.context_manager()
|
||||
.get_context(worker.job_id)
|
||||
.await
|
||||
.unwrap();
|
||||
.unwrap(); // safety: test
|
||||
assert_eq!(
|
||||
/* safety: test */
|
||||
ctx.state,
|
||||
JobState::Failed,
|
||||
"Iteration cap should transition to Failed, not Stuck"
|
||||
@@ -1989,4 +2062,52 @@ mod tests {
|
||||
"Should skip empty first reasoning and return the first non-empty one"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_store_fallback_in_metadata_roundtrip() {
|
||||
use crate::context::FallbackDeliverable;
|
||||
|
||||
let mut ctx = JobContext::new("Test", "fallback roundtrip");
|
||||
let memory = crate::context::Memory::new(ctx.job_id);
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "test failure");
|
||||
|
||||
// Store into metadata
|
||||
store_fallback_in_metadata(&mut ctx, Some(&fb));
|
||||
|
||||
// Verify it's stored and can be deserialized back
|
||||
let stored = ctx.metadata.get("fallback_deliverable");
|
||||
assert!(stored.is_some(), "fallback missing from metadata"); // safety: test
|
||||
|
||||
let recovered: FallbackDeliverable =
|
||||
serde_json::from_value(stored.unwrap().clone()).expect("deserialize fallback"); // safety: test
|
||||
assert_eq!(recovered.failure_reason, "test failure"); // safety: test
|
||||
assert!(!recovered.partial); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_store_fallback_handles_non_object_metadata() {
|
||||
use crate::context::FallbackDeliverable;
|
||||
|
||||
let mut ctx = JobContext::new("Test", "non-object metadata");
|
||||
ctx.metadata = serde_json::json!("not an object");
|
||||
|
||||
let memory = crate::context::Memory::new(ctx.job_id);
|
||||
let fb = FallbackDeliverable::build(&ctx, &memory, "failed");
|
||||
|
||||
store_fallback_in_metadata(&mut ctx, Some(&fb));
|
||||
|
||||
// Must normalize to object and store
|
||||
assert!(ctx.metadata.is_object()); // safety: test
|
||||
assert!(ctx.metadata.get("fallback_deliverable").is_some()); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_store_fallback_none_is_noop() {
|
||||
let mut ctx = JobContext::new("Test", "noop");
|
||||
let original = ctx.metadata.clone();
|
||||
|
||||
store_fallback_in_metadata(&mut ctx, None);
|
||||
|
||||
assert_eq!(ctx.metadata, original); // safety: test
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user