Prevent unsafe fallback and protect workflow journals

[beiwei30]

Summary

• suppress cross-provider fallback after streamed tool execution crosses the effect barrier
• interrupt and drain streamed tools while preserving failed-attempt usage and clean provider history
• serialize workflow journal appends and reject corrupt or duplicate resumes
• add deterministic provider and workflow concurrency regressions

Root cause

Provider retries treated any discardable partial stream as safe even after tool execution began. Workflow children opened journal.jsonl independently, while resume silently skipped malformed records. Together, these behaviors could duplicate externally visible tool effects or replay workflow steps after journal corruption.

Verification

• cargo fmt --all --check
• cargo clippy --workspace --all-targets -- -D warnings
• cargo check --workspace
• cargo test --workspace
• git diff --check

Notes

Includes a behavior-preserving ? rewrite in tools/src/web_cache.rs required by the strict Clippy gate.

[Copilot]

Pull request overview

This PR hardens streaming retry/fallback and workflow durability in orbcode’s core execution paths, aiming to prevent duplicated side effects (post-tool “effect barrier”) and to make workflow resume/journaling resilient to corruption and concurrent resumes.

Changes:

• Add an explicit “attempt discard disposition” so the retry layer can suppress cross-provider fallback once streamed tool execution has begun.
• Introduce cancellable streamed tool interruption/drain behavior and add regression tests covering pre-effect fallback vs. post-effect suppression.
• Serialize workflow journal appends and make workflow resume fail fast on corrupt journals or duplicate (already-active) run IDs.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
tools/src/web_cache.rs	Small clippy-driven refactor using ? in Option flow for cache lookup.
public-api-allow-list.txt	Allow-list the newly exported AttemptDiscardDisposition API.
model-provider/src/model.rs	Introduce AttemptDiscardDisposition and change ProviderStreamSink::discard_attempt to return it.
model-provider/src/lib.rs	Re-export AttemptDiscardDisposition from the crate’s public surface.
core/src/tool_flow.rs	Extend streamed tool execution tracking to support cooperative cancellation during interruption.
core/src/session_manager/tests/support.rs	Add/adjust test HTTP servers to simulate “thinking” errors vs. post-tool errors and to record fallback contact.
core/src/session_manager/tests/provider_streaming_retry.rs	Add deterministic regression tests for fallback suppression after tool side effects and for tool draining behavior.
core/src/session_manager/session_workflows.rs	Add a serialized journal writer and an in-process active-run reservation to prevent concurrent/duplicate workflow resumes.
core/src/session_manager/session_turn_loop.rs	Await streamed tool interruption to ensure tools are drained before proceeding.
core/src/session_manager/session_stream.rs	Plumb discard disposition from streamed-tool presence and await tool interruption during attempt discard.
core/src/session_manager/session_response.rs	Make streamed tool interruption async and introduce a dedicated cancellation flag for streamed tool execution.
core/src/retry.rs	Enforce “no fallback after tool execution started” based on AttemptDiscardDisposition.
core/src/lib.rs	Add new CoreError variants for active workflow runs and corrupt workflow journals.
core/src/compaction.rs	Update sink implementation to the new discard-attempt return type.

Suppressed comments (2)

core/src/session_manager/session_workflows.rs:278

• Drop should avoid panicking because it can lead to aborts during unwinding and makes shutdown paths fragile. Prefer recovering from a poisoned mutex (into_inner()) here instead of .expect(...).

        active_workflow_runs()
            .lock()
            .expect("active workflow registry mutex")
            .remove(&self.run_id);

core/src/session_manager/session_workflows.rs:269

• Mutex::lock().expect(...) can panic on a poisoned mutex. Since this is just a read check for whether a workflow is already active, it’s safer to recover the guard (into_inner()) and continue rather than crashing the process.

        active_workflow_runs()
            .lock()
            .expect("active workflow registry mutex")
            .contains(run_id)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.