feat(engine): observer/deny + sub-workflow-by-id + output_parser validation + cancellation + agent sub-ports (0.3.2→0.5.0)#3
Conversation
Add `run_with_checkpointer_journaled_observed` and `resume_with_checkpointer_journaled_observed`, which thread a host-supplied `Arc<dyn RunObserver>` through the durable + journaled path instead of hard-coding `NoopObserver`. The private `build_and_run` already threaded an observer end to end; `resume_with_checkpointer_inner` now takes one too. The existing non-observed entry points delegate to the observed variants with a `NoopObserver`, so behavior is unchanged for current callers. This unblocks live per-step run observation in downstream hosts (OpenHuman issue G2): a host observer now sees `on_step_finish` for every non-trigger node as a durable checkpointed run executes. Bump 0.3.0 -> 0.3.1 (additive, non-breaking). Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
… error port
Extend the durable resume seam so a host can *deny* a human-in-the-loop
approval gate, not just approve it. `resume_with_checkpointer_journaled_observed`
gains a `rejected: Vec<String>` parameter; denied gate ids are delivered to the
interrupted node(s) via a structured resume value `{ "rejected": [...] }`
(the bare `true` approval value is kept when nothing is denied, so the
approve-only path is unchanged). In `build_graph`, a gate whose id is in the
rejected set emits an error item on its `error` port when one is wired (so a
recovery sub-graph runs) or fails the run when it has none.
Bumps to 0.3.2.
Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
… docs truth-up Extend the `sub_workflow` node to run a child either from an inline `workflow` graph (unchanged, back-compat) OR a host-managed `workflow_id` reference. Since the engine is persistence-free, add a new host-injected `WorkflowResolver` capability (mirroring LlmProvider/ToolInvoker/…) that maps `workflow_id` → WorkflowGraph; the node calls `ctx.caps.resolver.resolve`. Cycle/depth handling: every nested sub_workflow run (inline or by id) increments a `run.sub_workflow_depth` counter threaded through a new `run_sub_workflow` entry point; a child exceeding MAX_SUB_WORKFLOW_DEPTH (8) is refused, bounding any indirect cycle (A→B→A). A direct self-reference (a resolved child that references the same workflow_id) is caught statically before it runs. validate() now requires exactly one of `workflow`/`workflow_id`. Docs truth-up: README + CLAUDE.md Status claimed jq/jaq expressions and retry backoff/timeouts were "pending" — they are implemented (src/expr.rs routes to jaq; engine.rs has fixed/exponential backoff + node_timeout_secs). Corrected. Adds engine API (new capability + Capabilities field), so bump 0.3.2 → 0.4.0. Mock resolver + tests: by-id resolves & executes, both-set/neither-set reject, self-reference reject, depth-limit enforced. cargo test green. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
…tive cancellation token, agent sub-ports
Phase 7 engine work (part B):
7.2 — output_parser now validates each input item against its configured
`config.schema` (a dependency-free JSON Schema subset: type/required/
properties/items/enum). On failure it makes one LLM auto-fix attempt via the
injected LlmProvider and re-validates; persistent failure surfaces a capability
error routed per the node's on_error policy. No schema ⇒ identity passthrough
(back-compat). Shared validate+fix routine lives in nodes/integration/schema.rs.
7.5 — new CancellationToken (Arc<AtomicBool>, runtime-agnostic) threaded through
the run loop and checked at each node boundary; a cancelled run stops scheduling
new node work and returns RunOutcome { cancelled: true }. New public entry points
run_cancellable / resume_cancellable; existing signatures unchanged (default
paths pass a fresh never-cancelled token). RunOutcome gains a `cancelled` field.
7.1 — agent node sub-ports (config-embedded): a `tools` sub-port surfaces
available tools to the model and honors a single-hop model-elected tool_call
(offered tools only), and an `output_parser` sub-port validates/repairs the
agent's output via the shared schema routine. Plain agent nodes are unchanged.
chat_model / memory sub-ports remain documented follow-ups (not stubbed).
Version 0.4.0 -> 0.5.0 (new public API + RunOutcome field). 260 lib tests green
(+21).
Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 40 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (12)
Comment |
# Conflicts: # src/engine.rs
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 01fe8b079e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let tool_conn = tool_call | ||
| .get("connection_ref") | ||
| .and_then(Value::as_str) | ||
| .or(conn); |
There was a problem hiding this comment.
When a model returns tool_call.connection_ref, this value is passed directly to ctx.caps.tools.invoke before falling back to the node's trusted config. In a prompt-injection or malformed-response scenario, an offered slug is enough for the model to choose an arbitrary host credential id for that tool call, so the credential should come from trusted config or the offered tool descriptor rather than from the LLM response.
Useful? React with 👍 / 👎.
| json!(true) | ||
| } else { | ||
| tracing::info!(?rejected, "resuming with denied approval gate(s)"); | ||
| json!({ "approved": newly_approved, "rejected": rejected }) |
There was a problem hiding this comment.
The structured resume value includes an approved list, but the gate handler only checks rejected and otherwise treats any ctx.resume.is_some() as approval. If a run is paused on multiple parallel approval gates and the host rejects or approves only one gate, every other interrupted gate not listed in rejected will execute instead of remaining pending, which can run user actions without the requested approval.
Useful? React with 👍 / 👎.
| return Ok(NodeResult::Update(items_update( | ||
| &node.id, | ||
| &[item], | ||
| Some("error"), | ||
| )?)); |
There was a problem hiding this comment.
When a denied gate has multiple recovery edges all leaving via the error port, fan_out_targets classifies the node as command-routed and the builder installs only command destinations instead of conditional edges. This branch returns a plain Update, so there is no conditional router to consume the recorded error port and none of the recovery branches are scheduled; use the fan-out command path for same-port error recovery.
Useful? React with 👍 / 👎.
…-D warnings) The observer parameter pushed resume_with_checkpointer_journaled_observed and resume_with_checkpointer_inner to 8 args; the SDK CI runs `cargo clippy --all-targets -- -D warnings`. Suppress consistently with the existing allow at engine.rs:851. No behavior change. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b7c39c4c5b
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| conn, | ||
| ) | ||
| .await?; | ||
| out.push(Item::new(validated)); |
There was a problem hiding this comment.
When output_parser has a schema, every validated input is wrapped with Item::new, which drops paired_item and binary metadata even when the JSON was already valid. In flows like split_out -> output_parser(schema) -> downstream, this loses the item pairing that downstream nodes use to correlate results with their originating input, while the no-schema passthrough path preserves it; clone the input item and replace only its json field after validation.
Useful? React with 👍 / 👎.
| pending_approvals, | ||
| // A cancelled token means at least one node boundary short-circuited, | ||
| // so surface the run as cancelled (its `output` is partial). | ||
| cancelled: token.is_cancelled(), |
There was a problem hiding this comment.
If callers cancel the token while the final node is already executing, that node can finish successfully and no later node boundary is skipped, but the outcome is still marked cancelled solely because the token is true at the end. This makes a complete terminal-node result look partial and can cause hosts to discard successful output; set this from a flag toggled in the cancellation short-circuit branch instead of sampling the token after execution.
Useful? React with 👍 / 👎.
Cargo.toml was bumped to 0.5.0 but Cargo.lock's tinyflows entry still read 0.4.0. Sync it so the lockfile matches the manifest. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
The agent tool sub-port took `tool_call.connection_ref` straight from the LLM response, so a prompt-injection or malformed model reply could select an arbitrary host credential id for the tool call. Credentials now come from trusted config only: the offered tool descriptor's `connection_ref`, falling back to the node's `connection_ref`. The model-supplied value is ignored. Adds regression tests asserting the offered descriptor's / node's connection wins over a hostile `tool_call.connection_ref`. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
…ex P1/P2) Two resume-path fixes flagged in review: 1. Keep unlisted approval gates pending (P1). tinyagents ignores the `with_update` state write on resume, so the resume value is the sole approval channel. The handler previously treated any `ctx.resume.is_some()` as approval, so resuming to approve/deny one gate would also run every other interrupted gate. Both resume paths now deliver the explicit `approved` gate-id list (checkpointed path also carries `rejected`), and a gate proceeds only if it is a bare-`true` resume (back-compat) or listed in `approved`; gates in neither stay pending. 2. Route denied fan-outs with a command (P2). When a denied gate's `error` port fans out to >=2 recovery nodes, the node is command-routed and has no conditional router; the denial branch emitted a plain (unrouted) update so no recovery branch ran. It now uses the `emit` fan-out command path, which drives all error-port successors (and still falls back to a plain update for the single/mixed-port case). Adds regression tests for both (a parallel-gate partial approval leaving the other gate pending, and a fan-out error-recovery denial). Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: fccd16b173
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // directly via a `Command::goto`; a single/mixed-port error | ||
| // edge falls back to a plain update the conditional-edge | ||
| // router consumes. | ||
| return Ok(emit(items_update(&node.id, &[item], Some("error"))?)); |
There was a problem hiding this comment.
When an approval gate has its normal main edge plus multiple recovery edges on the error port, fan_out is None because not all outgoing edges share one port, so this denial path emits only a plain update and leaves the duplicated error routes to the conditional router instead of scheduling both recovery nodes via Command::goto. In that common topology, denying the gate will not activate every error-recovery branch; route the matching error-port destinations explicitly rather than relying on the all-edges fan-out classification.
Useful? React with 👍 / 👎.
1 participant
Summary
Engine work driven by the OpenHuman tinyflows integration (host-side plan
docs/plans/tinyflows-integration). Four additive, back-compat commits taking the engine0.3.2 → 0.5.0. All existing entry points keep working; new capability/token/config are opt-in.296b59e) — newrun_with_checkpointer_journaled_observed/resume_with_checkpointer_journaled_observedvariants that thread aRunObserverend-to-end; existing entry points delegate viaNoopObserver. Lets a host persist per-step timing/status live instead of reconstructing post-hoc.96635d9) — a rejected approval gate routes to itserrorport (or fails the run if none) instead of being ignored.workflow_id(0bfa9e5) — thesub_workflownode now accepts EITHER the inlineworkflow(unchanged) OR aworkflow_idresolved via a new host-injectedWorkflowResolvercapability (6th field onCapabilities). Validation enforces exactly-one; direct self-reference is rejected statically and indirect cycles are bounded byMAX_SUB_WORKFLOW_DEPTH = 8. Plus docs truth-up: README/Roadmap no longer claim jq and retry/backoff are pending (both have long shipped —expr.rsroutes tojaq,engine.rshas fixed/exponential backoff +node_timeout_secs).01fe8b0):output_parsernow validates each item againstconfig.schema(a dependency-free JSON-Schema subset:type/required/properties/items/enum), attempts one LLM auto-fix on failure, re-validates, and routes persistent failures viaon_error. No schema ⇒ identity passthrough (back-compat).CancellationTokenchecked at node boundaries; newrun_cancellable/resume_cancellableentry points,RunOutcome.cancelled. Existing signatures unchanged (default = never-cancelled token).agentnode gains tool + output_parser sub-ports (config-embedded): tools surfaced to the model with single-hop execution of an electedtool_call; output post-processed through the shared validate+fix routine. Plain agent nodes are byte-for-byte unchanged.Compatibility / versioning
Capabilitiesgained a requiredresolverfield (breaking for direct constructors) → minor bump to 0.5.0.mock_capabilities()wires an empty resolver;mock_capabilities_with_resolver()added.run/resumeand inlinesub_workflowpaths are unchanged.Tests
cargo testgreen (260 passed, +40 over the base): observer threading, deny-on-resume routing, sub-workflow by-id resolution via a mock resolver, exactly-one/self-reference/depth-limit validation, output_parser valid/auto-fix/unfixable paths, cancellation before/after run, and agent tool/output_parser sub-ports.cargo test --features mock(reference-workflow e2e) green.Notes
feat/tinyflows-observer-deny(its two commits are the first two here).workflow_idlives host-side (OpenHumanWorkflowResolver).chat_model/memorysub-ports + multi-turn tool loop; the JSON-Schema subset could grow$ref/oneOf/numeric-bounds/pattern; wiring hostflows_cancel_runthrough the checkpointer entry points to the new token.Generated during the OpenHuman tinyflows integration (Phases 1, 7).