tandem-server 0.6.9

HTTP server for Tandem engine APIs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205

use super::*;

#[tokio::test]
async fn dead_letter_retry_honors_backoff_before_redispatch() {
    let state = test_state().await;
    let tenant = tenant("org-dl-backoff", "workspace-a", "operator-a");
    let scope = StatefulRuntimeScope::from_tenant_context(tenant.clone());
    let run_id = "run-dead-letter-backoff";
    let path = stateful_reliability_path_from_runtime_events_path(&state.runtime_events_path);

    let run = failed_run_with_snapshot(&state, run_id, tenant.clone(), "review").await;
    state
        .automation_v2_runs
        .write()
        .await
        .insert(run_id.to_string(), run);
    // Already dispatched far in the future -- its backoff window has not elapsed.
    let mut dead_letter = retry_requested_dead_letter("dead-backoff-effect", run_id, scope);
    dead_letter.status = StatefulDeadLetterStatus::Retrying;
    dead_letter.metadata = Some(json!({
        "retry_dispatched_at_ms": 9_000_000_000_000_u64,
        "retry_backoff_ms": 4_000,
    }));
    upsert_stateful_dead_letter(&path, dead_letter)
        .await
        .expect("seed retrying dead letter");

    let acted = state.dispatch_ready_stateful_dead_letter_retries().await;
    assert_eq!(
        acted, 0,
        "a dead letter inside its backoff window is skipped"
    );

    let run = state
        .get_automation_v2_run(run_id)
        .await
        .expect("run present");
    assert_eq!(run.status, AutomationRunStatus::Failed);
}

#[tokio::test]
async fn superseded_dead_letter_resolves_after_successful_replay() {
    let state = test_state().await;
    let tenant = tenant("org-dl-success", "workspace-a", "operator-a");
    let scope = StatefulRuntimeScope::from_tenant_context(tenant.clone());
    let run_id = "run-dead-letter-success";
    let path = stateful_reliability_path_from_runtime_events_path(&state.runtime_events_path);

    let run = failed_run_with_snapshot(&state, run_id, tenant.clone(), "review").await;
    state
        .automation_v2_runs
        .write()
        .await
        .insert(run_id.to_string(), run);
    let mut dead_letter = retry_requested_dead_letter("dead-success-effect", run_id, scope);
    dead_letter.status = StatefulDeadLetterStatus::Retrying;
    dead_letter.metadata = Some(json!({
        "superseded_by_success": true,
        "superseded_by_effect_id": "effect-success",
        "superseded_at_ms": 1_500,
    }));
    upsert_stateful_dead_letter(&path, dead_letter)
        .await
        .expect("seed superseded dead letter");

    let acted = state.dispatch_ready_stateful_dead_letter_retries().await;
    assert_eq!(acted, 1, "a superseded dead letter is resolved");

    let dead_letter = read_dead_letter(&state, &tenant, run_id, "dead-success-effect").await;
    assert_eq!(dead_letter.status, StatefulDeadLetterStatus::Resolved);
    assert_eq!(
        dead_letter.operator_disposition.as_deref(),
        Some("retry_succeeded")
    );
    // A resolved dead letter must not also re-drive the run.
    let run = state
        .get_automation_v2_run(run_id)
        .await
        .expect("run present");
    assert_eq!(run.status, AutomationRunStatus::Failed);
}

#[tokio::test]
async fn exhausted_dead_letter_is_parked_for_operator_review() {
    let state = test_state().await;
    let tenant = tenant("org-dl-exhausted", "workspace-a", "operator-a");
    let scope = StatefulRuntimeScope::from_tenant_context(tenant.clone());
    let run_id = "run-dead-letter-exhausted";
    let path = stateful_reliability_path_from_runtime_events_path(&state.runtime_events_path);

    let run = failed_run_with_snapshot(&state, run_id, tenant.clone(), "review").await;
    state
        .automation_v2_runs
        .write()
        .await
        .insert(run_id.to_string(), run);
    let mut dead_letter = retry_requested_dead_letter("dead-exhausted-effect", run_id, scope);
    dead_letter.status = StatefulDeadLetterStatus::Retrying;
    // Five dispatcher retries already recorded -- the cap counts these, not the
    // node/tool `attempts` field.
    dead_letter.metadata = Some(json!({ "retry_dispatch_count": 5 }));
    upsert_stateful_dead_letter(&path, dead_letter)
        .await
        .expect("seed exhausted dead letter");

    let acted = state.dispatch_ready_stateful_dead_letter_retries().await;
    assert_eq!(acted, 1, "an exhausted dead letter is parked");

    let dead_letter = read_dead_letter(&state, &tenant, run_id, "dead-exhausted-effect").await;
    assert_eq!(dead_letter.status, StatefulDeadLetterStatus::Ignored);
    assert_eq!(
        dead_letter.operator_disposition.as_deref(),
        Some("retry_exhausted")
    );
    // Exhausted retries must not re-drive the run.
    let run = state
        .get_automation_v2_run(run_id)
        .await
        .expect("run present");
    assert_eq!(run.status, AutomationRunStatus::Failed);
}

#[tokio::test]
async fn dead_letter_retry_skips_non_recoverable_run() {
    // A run that is already executing must never be re-driven out from under
    // its own executor by a dead-letter retry.
    let state = test_state().await;
    let tenant = tenant("org-dl-active", "workspace-a", "operator-a");
    let scope = StatefulRuntimeScope::from_tenant_context(tenant.clone());
    let run_id = "run-dead-letter-active";
    let path = stateful_reliability_path_from_runtime_events_path(&state.runtime_events_path);

    let mut run = failed_run_with_snapshot(&state, run_id, tenant.clone(), "review").await;
    run.status = AutomationRunStatus::Running;
    state
        .automation_v2_runs
        .write()
        .await
        .insert(run_id.to_string(), run);
    upsert_stateful_dead_letter(
        &path,
        retry_requested_dead_letter("dead-active-effect", run_id, scope),
    )
    .await
    .expect("seed dead letter for active run");

    let acted = state.dispatch_ready_stateful_dead_letter_retries().await;
    assert_eq!(acted, 0, "a running run must not be re-driven");

    let run = state
        .get_automation_v2_run(run_id)
        .await
        .expect("run present");
    assert_eq!(run.status, AutomationRunStatus::Running);
    let dead_letter = read_dead_letter(&state, &tenant, run_id, "dead-active-effect").await;
    assert_eq!(dead_letter.status, StatefulDeadLetterStatus::RetryRequested);
}

#[tokio::test]
async fn dead_letter_born_on_high_node_attempt_still_retries() {
    // Regression: `attempts` is the node/tool execution attempt at creation
    // (already at the executor cap for required-tool nodes). The retry cap must
    // count *dispatcher* retries, so the first requested retry must still
    // re-drive the run even when `attempts` is high.
    let state = test_state().await;
    let tenant = tenant("org-dl-high-attempt", "workspace-a", "operator-a");
    let scope = StatefulRuntimeScope::from_tenant_context(tenant.clone());
    let run_id = "run-dead-letter-high-attempt";
    let path = stateful_reliability_path_from_runtime_events_path(&state.runtime_events_path);

    let run = failed_run_with_snapshot(&state, run_id, tenant.clone(), "review").await;
    state
        .automation_v2_runs
        .write()
        .await
        .insert(run_id.to_string(), run);
    let mut dead_letter = retry_requested_dead_letter("dead-high-attempt-effect", run_id, scope);
    // Node/tool attempts already at the executor cap -- must NOT pre-exhaust.
    dead_letter.attempts = 5;
    upsert_stateful_dead_letter(&path, dead_letter)
        .await
        .expect("seed dead letter");

    let acted = state.dispatch_ready_stateful_dead_letter_retries().await;
    assert_eq!(
        acted, 1,
        "a high node-attempt count must not block the first dispatcher retry"
    );

    let run = state
        .get_automation_v2_run(run_id)
        .await
        .expect("run present");
    assert_eq!(run.status, AutomationRunStatus::Queued);
    let dead_letter = read_dead_letter(&state, &tenant, run_id, "dead-high-attempt-effect").await;
    assert_eq!(dead_letter.status, StatefulDeadLetterStatus::Retrying);
    assert_eq!(
        dead_letter
            .metadata
            .as_ref()
            .and_then(|meta| meta.get("retry_dispatch_count"))
            .and_then(|value| value.as_u64()),
        Some(1)
    );
}