feat(cp): OAB runtime CP client — [control_plane] config, worker serving, headless mode (PR 3/4) - #1471
feat(cp): OAB runtime CP client — [control_plane] config, worker serving, headless mode (PR 3/4)#1471chaodu-agent wants to merge 8 commits into
Conversation
This comment has been minimized.
This comment has been minimized.
2330bba to
3f3d181
Compare
3e246c0 to
9dc41e7
Compare
This comment has been minimized.
This comment has been minimized.
3f3d181 to
4a290d3
Compare
9dc41e7 to
c2030f2
Compare
This comment has been minimized.
This comment has been minimized.
4a290d3 to
b23b309
Compare
c2030f2 to
995b16c
Compare
This comment has been minimized.
This comment has been minimized.
github-actions
Bot
added
pending-maintainer
pending-contributor
and removed
pending-contributor
pending-maintainer
labels
b23b309 to
718a9f8
Compare
995b16c to
68dbe7e
Compare
This comment has been minimized.
This comment has been minimized.
github-actions
Bot
added
pending-maintainer
pending-contributor
and removed
pending-contributor
pending-maintainer
labels
718a9f8 to
08fd30c
Compare
68dbe7e to
a120ea6
Compare
This comment has been minimized.
This comment has been minimized.
08fd30c to
7473568
Compare
This comment has been minimized.
This comment has been minimized.
…teardown Round-7 review fixes: - F1: failed() now caps every error string (64 KiB, UTF-8-safe) inside the constructor so no call site can bypass it; oversized-error regression test mirrors the success-path one - F34: delegation_session_key mixes in the CP admission token, so a re-admission of a reusable id can never resume an earlier admission's session (e.g. one orphaned by a drain-timeout abort); false reconnect claim in the doc corrected; key-property test pins admission scoping - F22/F31: duplicated cancel+discard blocks extracted into bounded helpers; discard now shares the 5s bound instead of being unbounded on every terminal path - F15: client WS now sets max_message_size/max_frame_size (1 MiB), mirroring the CP's accept-side max_frame_bytes instead of tungstenite's 64 MiB default - F5: the cp/register ack wait is bounded (10s, mirroring the CP's register_timeout_secs) so a stalled CP enters backoff instead of hanging - F27: config/docs no longer claim a local min() the code does not implement — the runtime enforces the ack value - F37: ADR section 7 facade label corrected from PR 3/4 to PR 4/4
This comment has been minimized.
This comment has been minimized.
github-actions
Bot
added
pending-maintainer
pending-contributor
closing-soon
|
Caution This PR has been waiting on the author for more than 2 days (labeled @chaodu-agent — You must add a new comment on this PR to remove the |
|
🔒 Auto-closing: this PR has had the If you'd like to continue working on this, please submit a new PR and link to this one if necessary. |
|
reopened |
…er success discard Round-8 review fixes: - F46: cap_text and cap_result now evaluate JSON-escaped byte length instead of raw UTF-8 length so control-character and quote/slash escape inflation cannot inflate a 512 KiB result past the CP's pre-parse 1 MiB max_frame_bytes limit; regression test pins escape-heavy and quote-heavy inflation safety; client::send() adds an outbound transport size guard - F52: cancel_and_discard() now shares a single 5s deadline across both the cancel and discard phases (discard uses whatever budget remains, or is skipped if cancel consumed the bound), keeping total teardown strictly <= DRAIN_TIMEOUT (5s); regression test pins stalled cancel + stalled discard bound - F53: success path now spawns session discard in the background off the critical path, preventing a slow or wedged discard from delaying result delivery to the initiator; safe because session keys are admission-scoped - F54: executor module-level invariant doc updated to state the three-parameter session key (instance_id, delegation_id, admission) - F55: truncation marker uses ASCII "..." instead of U+2026 ellipsis - F51: documented MAX_INBOUND_FRAME_BYTES relation to CP-side configurable max_frame_bytes and future RegisterAck negotiation direction
Round-8 Review Fixes (Head
|
openab-app
Bot
removed
the
closing-soon
Self-Review & CI Verification Summary (Head
|
Independent Deep Systems Review (Model: Gemini Pro Extended Thinking)An adversarial systems-level code review of branch 1. High-Confidence Invariant Verifications (Passed)
2. Architectural Observations & Follow-Up Recommendations
VerdictLGTM for PR #1471 scope. The Round 8 fixes (F46, F51-F55) are verified correct, CI is 100% green at |
…dless doc (review nits)
|
Important CHANGES REQUESTED What This PR DoesThis stacked PR connects the OAB runtime to the Agent Control Plane. An optional How It Works
Findings
Finding Details🟡 F1: Require an explicit secure transport policyThe credential is the complete agent identity and is inserted into the Requested change: reject non-loopback 🟡 F2 + F3: Make reconnect behavior fleet-safe
Requested change: add a shutdown-aware connect timeout and full/equal jitter. Add paused-time tests for the timeout, ceiling, reset threshold, and jitter range. 🟡 F4: Do not let lock poisoning become process abort
Requested change: use an unpoisoned mutex or recover with 🟡 F5 + F6: Validate both sides of the protocol contractThe client registers its role and protocol version but then accepts every Requested change: reject 🟡 F7: Redact the credential at the owning typeHeader sensitivity protects only request formatting. The source config remains a normal Requested change: implement redacted 🟡 F8: Make the headless contract describe behaviorThe code intentionally gives primary + Requested change: rename the variant to behavior-oriented 🟡 F9: Name the shared frame contract accuratelyThe latest outbound guard correctly uses the same 1 MiB ceiling as inbound WebSocket configuration, but Requested change: rename it to 🟡 F10: Supervise the control-plane taskThe handle is stored until shutdown. If Requested change: supervise the handle and log/fail readiness on early exit; in worker-only mode, consider terminating so the deployment supervisor can recover it. 🟡 F11: Keep startup remediation completeThe startup condition recognizes LINE, LINE WORKS, and Teams through Requested change: include Baseline Check
What's Good (🟢)
Addressing External Feedback
Group Review Outcome
5. Three Reasons We Might Not Need This PR
The ADR already chose an in-process client, so these are architecture/timing tradeoffs rather than separate blockers. The requested changes above make that chosen model safer and easier to operate. |
![chaodu-obk[bot]](https://avatars.githubusercontent.com/u/262003297?s=60&v=4){kind=small}
There was a problem hiding this comment.
Important
CHANGES REQUESTED
Consolidated review: #1471 (comment)
GitHub event: COMMENT -- self-review delivery only; this is not an approval.
| #[derive(Debug, Clone, Deserialize)] | ||
| #[serde(deny_unknown_fields)] | ||
| pub struct ControlPlaneConfig { | ||
| /// CP WebSocket endpoint, e.g. `wss://cp.internal:9800/cp`. The server |
There was a problem hiding this comment.
🟡 F1 - Require an explicit secure transport policy
auth_key is the complete bearer credential, but this endpoint accepts non-loopback ws:// without rejection or an explicit insecure opt-in. That sends the credential in cleartext over the network.
Requested change: reject non-loopback ws:// by default, or require an explicit allow_insecure_transport opt-in with a startup warning and documented local-development exception.
| ..Default::default() | ||
| }; | ||
| let (ws, _resp) = | ||
| tokio_tungstenite::connect_async_with_config(request, Some(ws_config), false) |
There was a problem hiding this comment.
🟡 F2 - Bound the complete dial
connect_async_with_config covers DNS, TCP, TLS, and the WebSocket handshake but has no explicit deadline. A blackholed endpoint can therefore stall this reconnect iteration until the OS timeout, far beyond the documented retry schedule.
Requested change: wrap the dial in a shutdown-aware timeout and add a paused-time regression test.
| return; | ||
| } | ||
| } | ||
| backoff = (backoff * 2).min(MAX_BACKOFF_SECS); |
There was a problem hiding this comment.
🟡 F3 - Jitter fleet reconnects
Every replica follows the same deterministic 1/2/4/8/16/30-second sequence. After a CP restart, the fleet can reconnect in lockstep and create a thundering herd.
Requested change: apply full or equal jitter while preserving the ceiling, and test the allowed delay range.
|
|
||
| /// Number of admitted, not-yet-finished delegations. | ||
| pub fn active(&self) -> u32 { | ||
| self.inflight.lock().expect("inflight mutex").len() as u32 |
There was a problem hiding this comment.
🟡 F4 - Recover this bookkeeping lock from poisoning
All inflight operations use expect("inflight mutex"). If any panic poisons the lock, SlotGuard::drop can panic again while unwinding through release, causing a process abort. The map stores only tokens and Arc<Notify>, so recovery is safe.
Requested change: use an unpoisoned mutex or recover with PoisonError::into_inner() consistently, including the drop path, and add a poison-recovery test.
| }; | ||
|
|
||
| match method.as_str() { | ||
| methods::DELEGATE => { |
There was a problem hiding this comment.
🟡 F5 - Fail closed on the registered role
This branch accepts cp/delegate for both Worker and Primary. That makes hub routing the only role boundary even though the runtime already has the configured role.
Requested change: reject delegation requests unless self.cfg.agent_type == CpAgentType::Worker, and test the primary path.
| /// **Strict.** An unknown key here is a hard startup failure rather than a | ||
| /// silently-defaulted one: a mistyped `max_delegated_sessions` would otherwise | ||
| /// look effective while the runtime advertised the default budget of 1. | ||
| #[derive(Debug, Clone, Deserialize)] |
There was a problem hiding this comment.
🟡 F7 - Redact the credential at its owning type
Deriving Debug includes auth_key verbatim. Header sensitivity protects request formatting only; one future config diagnostic can expose the bearer credential.
Requested change: provide a manual redacted Debug implementation or use a secret wrapper whose Debug is always masked.
| // (visible in the roster, ready for primary-side initiation in the | ||
| // next slice). Facade-only forecloses the client entirely. | ||
| if cfg.control_plane.is_some() { | ||
| HeadlessMode::ControlPlaneWorker |
There was a problem hiding this comment.
🟡 F8 - Name and document the actual full-boot mode
This ControlPlaneWorker result also represents type = "primary" plus [mcp]. The config table still omits that combination and leaves the primary/no-adapter error row unqualified.
Requested change: rename the variant to behavior-oriented FullBoot/ControlPlaneClient, add the primary + [mcp] row, and qualify the error row as primary without [mcp].
|
|
||
| async fn send(sink: &mut WsSink, frame: &JsonRpcRequest) -> anyhow::Result<()> { | ||
| let text = serde_json::to_string(frame)?; | ||
| if text.len() > MAX_INBOUND_FRAME_BYTES { |
There was a problem hiding this comment.
🟡 F9 - Name the bidirectional frame contract accurately
The complete outbound frame is now checked against MAX_INBOUND_FRAME_BYTES, while the constant comment says outbound is already capped elsewhere and this closes only the inbound direction. Payload-field caps and complete-frame caps are distinct.
Requested change: rename this to MAX_FRAME_BYTES (or split the limits) and document its use for both WebSocket receive configuration and serialized sends.
| runner, | ||
| std::time::Duration::from_secs(prompt_hard_timeout_secs), | ||
| )); | ||
| tokio::spawn(client.run(shutdown_rx.clone())) |
There was a problem hiding this comment.
🟡 F10 - Supervise the control-plane task during normal operation
This handle is only awaited during shutdown. If run returns early or panics, the process can remain healthy-looking while no longer registering or serving; a worker-only process has no other duty that exposes the failure.
Requested change: supervise early task exit and fail readiness or terminate worker-only mode so the deployment supervisor can recover it.
| } | ||
| HeadlessMode::None => { | ||
| anyhow::bail!( | ||
| "no adapter configured — add [discord], [slack], [telegram], [wecom], [googlechat], or [gateway] to config (or [mcp] for facade-only mode, or [control_plane] with type = \"worker\" for control-plane worker mode), or set platform env vars (TELEGRAM_BOT_TOKEN, etc.)" |
There was a problem hiding this comment.
🟡 F11 - Keep startup remediation complete
has_unified_platform recognizes LINE, LINE WORKS, and Teams, but this error omits [line], [lineworks], and [teams] from the valid adapter list.
Requested change: include those adapters or replace the brittle enumeration with a reference to the platform configuration documentation.
2 participants
OAB runtime CP client — [control_plane] config, worker serving, headless mode (PR 3/4)
Implements ADR §3 (Registration, OAB side), §3-Headless, and the worker half of §4 —
docs/adr/agent-control-plane.md.Discord Discussion URL: https://discord.com/channels/1491295327620169908/1491365158868619404/1532377620241449040
Review Contract
Goal
An OAB runtime can join the control plane: a new optional
[control_plane]config section makes the runtime dial the CP over WebSocket, register with its key-bound identity, heartbeat, and — whentype = "worker"— serve incomingcp/delegaterequests by running the prompt through its local ACP session pool and replyingcp/delegate_resultwithin the deadline.type = "worker"also unlocks headless mode:[agent]+[control_plane]with no platform adapters is a valid boot.Non-goals
spawn_agent, …) — PR 4/4 (ADR §6).openab agentCLI — PR 4/4.target_disconnected(ADR §4 v1 contract).Accepted Residual Risks
target_disconnectedsynthesis is the single source of truth. Deliberate: per-connection ownership is the ADR's replica-safety rule.stream_prompt_blocks→PromptExecution) touches the shared platform delivery path; safety is argued by mechanical mapping + every pre-existing adapter/dispatch test passing unmodified, not by new platform-level tests.x86_64-pc-windows-gnutarget); no platform-specific code was added and the crate compiles with--no-default-features.Acceptance Criteria
[control_plane]→ zero behavior change (config parse test:None; no task spawned)[agent]+[control_plane type="worker"], no adapters → boots headless;type="primary"alone without adapters still bails;[mcp]-only unchanged;[mcp]+[control_plane]runs facade AND client (run-mode test matrix)openab-cpserver in-process (integration test, not a mock): register/ack, heartbeat keeps lease, delegate→result roundtrip, cancel mid-flight, local deadline →Timeout, lease-expiry close → reconnect + re-register succeedsinstance_idper process across reconnectsdelegation_id→ immediateFailed, never executed, CP capacity released); outcome mapping Completed/Failed/Timeout/Cancelled pinned by unit tests; cancel/deadline cleanup bounded (5s) and the pool session is discarded — no slot leaksCompletedresult bodies are capped (512 KiB, UTF-8-safe marker) below the CP'smax_frame_bytestransport limit — an oversized agent result can no longer drop the connection and kill co-inflight delegationsopenab-cpgains a defaultserverfeature;openab-coredepends on itdefault-features = false—cargo treeshows no axum edge into the runtime; theopenab-cpbinary still builds with default features; zero Dockerfile changes neededauth_keynever enters the agent child env (untouchedenv_cleardiscipline) and never appears in logsFollow-ups
wss://…/acp; the server mounts/cp— example corrected in this PR; revisit if an alias is preferred instead.At a Glance
Prior Art & Industry Research
crates/openab-core/src/gateway.rs).[mcp](feat(mcp): facade-only run mode — adapter-less [mcp] config is valid #1453).AdapterRouter+SessionPool), rather than the ACP-over-WSacp_clientsynthesis path — one seam, no event fabrication.Proposed Solution
crates/openab-cp): defaultserverfeature gates axum/registry/router/policy/events/server + the binary;proto(wire types) stays unconditional. The runtime consumesopenab-cpwithdefault-features = false— wire types only, no server deps, no new Dockerfile stubs.crates/openab-core/src/config.rs):ControlPlaneConfigwithdeny_unknown_fields,CpAgentType(primary|worker—observerunrepresentable by construction),${ENV}expansion free via the existing pass, validation for empty fields andmax_delegated_sessions > 0.crates/openab-core/src/control_plane/client.rs): process-lifetimeinstance_id; Bearer auth at upgrade;cp/registerfirst frame; single select! loop over inbound / heartbeat / completion channel / shutdown; on disconnect cancels local work, drains (5s) then aborts, backs off shutdown-aware.crates/openab-core/src/control_plane/executor.rs): admission (cap, duplicate) decided before execution; fresh ACP session per delegation keyedcontrol-plane:<sha256(instance_id + delegation_id)>; budgetmin(deadline − now, prompt_hard_timeout);PromptRunnertrait seam so integration tests inject a fake runner while production uses the pool.crates/openab-core/src/adapter.rs):stream_prompt_blocksreturnsPromptExecution { final_text, terminal_error, silent_failure }; platform callers map back mechanically — no observable change, pre-existing tests untouched.src/main.rs): client spawned after router construction when the section is present; stopped (or aborted after 10s) beforepool.shutdown(); headless run-mode matrix extended.Alternatives Considered
openab-cp-clientcrate: rejected — every one of the 19 Dockerfiles stubs workspace crates and would need edits; a module insideopenab-coreis Dockerfile-neutral and the client is small.acp_clientevent-synthesis path for delegated prompts: rejected — it fabricates gateway events to re-enter the dispatcher; the cron precedent calls the router/pool seam directly with less indirection and no fake sender identity.Failedso CP capacity releases.Validation
cargo test -p openab-cp— 94 passed (server unaffected by the feature split; binary still builds)cargo test -p openab-core— 713 passed, +36 new (config/executor/client incl. the transport-cap regression); 1 failure is the known pre-existing macOS-onlysecrets::tests::resolve_exec_nonzero_exit, present at baselinecargo test -p openab-core --test cp_client— 7 integration tests against the real in-process CP server: registration, heartbeat/lease, roundtrip, cancel, timeout, lease-expiry reconnect, shutdown deregistrationcargo test(root bin) — 25 passed incl. the headless run-mode matrixcargo clippy --workspace --all-targets— warning count identical to baseline (12);cargo fmt --checkdiff count identical to baseline (217, pre-existing drift untouched)cargo check --workspace --no-default-features— passescargo check --target x86_64-pc-windows-gnu— UNVERIFIED (target not installed on the build host); no platform-specific code added