From fd02ef2aed4e6f00a3f4d13e9fe8cc7b7e07fe40 Mon Sep 17 00:00:00 2001 From: Willow Sparks Date: Sun, 1 Nov 2026 01:22:07 +0000 Subject: [PATCH] fix(harness): restore committed refinement presentation The generic deferred-job outcome path hid the existing refinement decoration. Let successful plugins project the existing custom-message payload through result.message; the daemon remains unaware of harness policy. Harness Lua builds refinement_outcome only after atomic apply using actual committed before/after values, preserving applied and summary results. Invalid projections show a visible diagnostic with the generic outcome without changing committed terminal state; failed jobs remain generic errors. Native rustfmt and git diff --check pass. Focused core committed create/update/delete/empty projection tests, daemon generic/projected/malformed/failed transport tests and CLI real-apply-to-TUI-decoder assertion are authored but native execution is explicitly pending the shared build slot. No settings, default constants, live runtime, dependency direction or TUI schema changes. --- .../pa-cli/tests/plugin_headless/harness.rs | 7 ++ crates/pa-core/README.md | 6 +- crates/pa-core/src/harness_plugin/policy.lua | 16 +++- crates/pa-core/src/harness_plugin/tests.rs | 60 +++++++++++++++ crates/pa-daemon/README.md | 9 ++- crates/pa-daemon/src/plugin_commands.rs | 76 +++++++++++++------ crates/pa-daemon/src/plugin_job_events.rs | 21 +++++ 7 files changed, 166 insertions(+), 29 deletions(-) diff --git a/crates/pa-cli/tests/plugin_headless/harness.rs b/crates/pa-cli/tests/plugin_headless/harness.rs index 12c3bd64b..c71b5303d 100644 --- a/crates/pa-cli/tests/plugin_headless/harness.rs +++ b/crates/pa-cli/tests/plugin_headless/harness.rs @@ -226,6 +226,13 @@ fn sparse_automatic_settings_publish_refinement_before_cli_exit() { assert_eq!(jobs.len(), 1, "{events:#?}"); assert_eq!(jobs[0]["job"]["state"]["status"], "completed", "{jobs:#?}"); assert_eq!(jobs[0]["job"]["state"]["result"]["summary"], "CLI learned"); + let message = &jobs[0]["job"]["state"]["result"]["message"]; + let rendered = pa_tui::custom_message::custom_message_entries(message); + assert!( + matches!(rendered.as_slice(), [pa_tui::chat::ChatEntry::RefinementOutcome(row)] + if row.summary == "CLI learned" && row.edits.len() == 1), + "{rendered:?}" + ); assert!( std::fs::read_to_string(fixture.agent.join("plugins/harness/store.json")) .unwrap() diff --git a/crates/pa-core/README.md b/crates/pa-core/README.md index 1b9ce175c..19e26d7e7 100644 --- a/crates/pa-core/README.md +++ b/crates/pa-core/README.md @@ -80,7 +80,11 @@ No provider HTTP (pa-ai), no loop policy (pa-agent), no daemon supervision (pa-d `session_engine::PluginJobSink` and `AgentSession::set_plugin_job_sink` let hosts observe future terminal deferred-job outcomes. Callbacks run outside the job registry lock; retired generations cannot publish. Observation never consumes -results returned by the independent drain API. +results returned by the independent drain API. Successful job results may include +`message` using the existing custom-message payload (`customType`, `content`, +`display`, optional object `details`) for a host to present. The optional harness +plugin builds its refinement outcome only after atomic store apply, with committed +before/after values; the generic job registry does not interpret that payload. `SessionEngine` (message in -> events out), `export_html` (the HTML exporter plus its renderer seam: `tools_section`, `pre_render_custom_tools` + `ToolHtmlRenderer`/`RenderedToolHtml`/`RenderedToolResult`, and `ansi_to_html::ansi_lines_to_html`/`tool_render::trim_rendered_result_lines` for renderer implementers), `ToolRegistry`, kernel manager, settings (`SettingsManager`), packages (`packages::PackageManager` + source types), `platform` (process control, file locking, permissions, shell selection - usable by higher crates, e.g. pa-cli's detached spawn), `session_engine::rlm_host::RlmSubagentHost` (implemented by pa-daemon for supervisor-backed children and by pa-cli's print mode for in-process children; the default is no children; the trait crate also owns the shared spawn/depth facts — `DEFAULT_RLM_MAX_DEPTH`, the depth-limit and name errors, the daemonless `create_session` refusal, `rlm_child_session_dir`, and the `persisted_rlm_max_depth`/`resolve_rlm_max_depth` precedence chain), `session_engine::semantic_edges` (the TS `semantic-edges.ts` WRITER: `SemanticEdgeRecorder`/`SemanticEdgeIdentity`, `semantic_edge_ledger_path` + `last_committed_request_id`, and the `X-ACP-Model-Request-ID`/`Idempotency-Key` constants — one durable per-session ledger of model-request events, byte-compatible with the TS lines; the fold and any publishing stay downstream), `mcp::McpManager` (the session's host-side MCP manager, exposed as `SessionEngine.mcp_manager`: auth gating for built-in integrations, the `mcpServers` setting seam, and the `mcp.*` host-request handlers - `mcp.config`/`mcp.refresh`/optional `mcp.begin_login` - the kernel's generic MCP client resolves through), `mcp::McpManager::connection_roster` + `mcp::McpConnectionEntry` (the `/mcp` view's roster: every resolved integration with its connected state, display kind, transport, and generic-listable flag), `mcp::McpManager::service_catalog_views` + `service_catalog_diagnostics` (the `/mcp` view's resolved service-catalog cards and their diagnostics, answered from local state - the TS `buildServiceCatalogViews` parity; the picker never round-trips the kernel), `mcp::McpManager::api_key_credential_views` + `mcp::API_KEY_CREDENTIALS`/`McpCredentialView` (the `/mcp` view's api-key credential rows - the stored keys the surface manages alongside the connections, e.g. the web-search key - configured from the shared auth store), `models` (the registry and resolver, plus `models::order_for_picker`: the picker-catalog order the composition root applies to its snapshot), `session_engine::provider_adapter` (`switchable_stream_fn`/`ProviderTarget`: the live provider-target slot a host swaps on `set_model`; the adapter bridges the loop's abort into pa-ai's `StreamOptions::signal` cancellation token, so a turn abort cancels the in-flight HTTP fetch at the transport — closed by `ModelStream::close` and the stream's drop), `session_engine::ipython_state` (the post-compaction `ipython_state` kernel-persistence notice: the `CompactionKernelProbe` view of the kernel provisioner the engine wiring binds via `AgentSession::set_kernel_state_probe`, and the notice row that rides `CompactRun` for the surfaces to broadcast). `session_engine::session_commands` (the `/compact`/`/refine`/`/goal`/`/autonomous` executor: the durable echo/result rows with their persistence order - the echo ahead of the command so its own work sees it - the compaction/refinement/goal/autonomous arms, and the post-execution live-context rebuild that keeps the rows in the agent state; hosts own the wire surface), `session_engine::auto_refine_trigger` (the compact-trigger auto-refine machine on `AgentSession`: the arm/discard/increment accessors and `consume_compact_auto_refine`, the shared gate/review/stamp sequence every scheduling surface applies - TS `_compactAutoRefinePending`/`_lastAutoRefineReviewAt`/`_assistantTurnsSinceAutoRefine` plus `_maybeAutoRefine`'s compact arm, the serialized checkpoint's compact step, and dispose's serialized drain, with the surface parameter carrying the one behavioral difference between a boundary and a disposal). `session_engine::goal_boundary` (the goal arms of the natural turn end on `SessionEngine`: the `--goal` construction seed with its next-turn context row - TS `_pendingNextTurnMessages` on `AgentSession` - the usage accounting, the budget-limit steer mint, the natural continuation mint with its slot rollback, the RLM deferral behind unsettled descendant work (mark/drop/take owed, the wake-up ownership probe, and the latest settled turn the post-settle deliveries decide from), and the terminal-error finish; the transports - the print driver's in-loop hook, the ACP settle loop, the daemon worker's queue - drive the one goal driver through these methods). `SessionEngine::expand_skill_submission` + `AgentSession`'s prompt-path expansion (the `/skill:` submission seam, TS `_expandSkillCommand`: `skills::expand_skill_command` builds the `` block over the shared `pa_types::skill_blocks` parse, the accepted-turn row and the admitted user message agree on the expanded text, and the `skill_use_count` session counter counts at the admission). `workspace_snapshot` (`create_workspace_snapshot`: bounded, secret/nested-repo/submodule/symlink-ancestor-excluding capture of a git worktree into an owner-private staging directory, with the caller-chosen `BaselineMode` (HEAD-tree content verified against HEAD's object ids, or explicit external) and a `ConcurrentMutation` error when the worktree changes under the capture; `verify_workspace_snapshot`: offline manifest/blob/baseline integrity check; the `SnapshotManifest`/`WorkspaceSnapshot`/`SnapshotLimits`/`BaselineMode`/`SnapshotError` types). All subsystem internals `pub(crate)`; the engine is the only facade - subsystems must not import each other directly (the package manager consumes the public settings API only). diff --git a/crates/pa-core/src/harness_plugin/policy.lua b/crates/pa-core/src/harness_plugin/policy.lua index 8947a3838..31320547c 100644 --- a/crates/pa-core/src/harness_plugin/policy.lua +++ b/crates/pa-core/src/harness_plugin/policy.lua @@ -239,8 +239,22 @@ local function refinement_job(input,ctx) input={history=snapshot.history,records=records,instructions=input.instructions,review=review}}) local proposal=ctx.call("native.json.decode",{text=plan.text}) assert(type(proposal.edits)=="table", "proposal needs edits") - return ctx.call("harness.store.apply",{scope=target,baseline=records,proposal=proposal, + local result=ctx.call("harness.store.apply",{scope=target,baseline=records,proposal=proposal, branch=snapshot.branch,reason=input.reason}) + -- Presentation follows committed changes, never the model's proposed state. + local edits=proposal.edits + for index,change in ipairs(result.applied) do + local proposed=proposal.edits[index] + local edit={action=proposed.action,kind=proposed.kind,id=proposed.id,applied=true} + if change.previous.state=="Present" then edit.before=change.previous.value end + if change.current.state=="Present" then edit.after=change.current.value end + edits[index]=edit + end + local summary=type(result.summary)=="string" and result.summary or "" + result.message={customType="refinement_outcome",content=summary, + display=true,details={summary=summary, + scope=target=="global" and "global" or "local",edits=edits}} + return result end local function history(input,ctx) return ctx.call("harness.store.history",{identity=identity(input,ctx)}) diff --git a/crates/pa-core/src/harness_plugin/tests.rs b/crates/pa-core/src/harness_plugin/tests.rs index 7062c9591..59fd5955c 100644 --- a/crates/pa-core/src/harness_plugin/tests.rs +++ b/crates/pa-core/src/harness_plugin/tests.rs @@ -227,6 +227,20 @@ async fn sparse_automatic_settings_review_plan_apply_and_publish_completion() { let terminal = f.engine.drain_plugin_jobs().await; assert_eq!(terminal.len(), 1, "{terminal:?}"); assert_eq!(terminal[0]["state"]["status"], "completed", "{terminal:?}"); + assert_eq!( + terminal[0]["state"]["result"]["message"], + json!({ + "customType":"refinement_outcome", "content":"fixture learned", "display":true, + "details":{"summary":"fixture learned","scope":"local","edits":[{ + "action":"create","kind":"memory","id":"learned","applied":true, + "after":{"content":"auxiliary route"} + }]} + }) + ); + assert_eq!( + terminal[0]["state"]["result"]["applied"][0]["current"], + json!({"state":"Present","value":{"content":"auxiliary route"}}) + ); assert_eq!( f.call("get", json!({"kind":"memory","id":"learned"})) .await @@ -281,6 +295,52 @@ async fn await_job(f: &Fixture, id: &Value) -> Value { .await .expect("deferred job settled") } +#[tokio::test] +async fn refinement_presentation_uses_committed_update_delete_and_empty_changes() { + let original = r#"{"summary":"fixture learned","edits":[{"action":"create","kind":"memory","id":"learned","value":{"content":"auxiliary route"}}]}"#; + for action in ["update", "delete", "none"] { + let edits = if action == "none" { + json!([]) + } else { + json!([{"action":action,"kind":"memory","id":"learned","value":{"content":"changed"}}]) + }; + let proposal = json!({"summary":"fixture learned","edits":edits}); + let mut config = settings(true); + config["auxiliaryModel"] = json!("aux/selected"); + config["testProvider"] = json!(AUX_PROVIDER.replace(original, &proposal.to_string())); + config["plugins"] = json!({"route":{"enabled":true,"path":"provider.lua"}}); + let f = Fixture::new(config, LEGACY).await; + f.call( + "create", + json!({"kind":"memory","id":"learned","value":{"content":"before"}}), + ) + .await + .unwrap(); + let job = f.call("refine", json!({})).await.unwrap(); + let result = await_job(&f, &job["id"]).await; + assert_eq!(result["status"], "completed", "{result}"); + let edits = result["result"]["message"]["details"]["edits"] + .as_array() + .unwrap(); + if action == "none" { + assert!(edits.is_empty()); + } else { + assert_eq!(edits[0]["action"], action); + assert_eq!(edits[0]["before"], json!({"content":"before"})); + assert_eq!(edits[0]["applied"], true); + if action == "update" { + assert_eq!(edits[0]["after"], json!({"content":"changed"})); + } else { + assert!(edits[0].get("after").is_none()); + } + } + assert_eq!( + result["result"]["applied"].as_array().unwrap().len(), + edits.len() + ); + } +} + #[tokio::test] async fn manual_and_interval_refinement_use_scoped_explicit_auxiliary_review_and_plan() { let mut settings = settings(true); diff --git a/crates/pa-daemon/README.md b/crates/pa-daemon/README.md index 2585d7598..f21b954e7 100644 --- a/crates/pa-daemon/README.md +++ b/crates/pa-daemon/README.md @@ -278,8 +278,13 @@ No agent behavior inside workers beyond hosting a pa-core engine; no UI. `AgentSessionEngine::set_plugin_job_sink` wires the host observer onto current and future sessions. The worker persists terminal generic plugin-job outcomes and publishes them through the existing custom-message event transport, fenced by -the durable session identity. Failed outcome persistence is visible without -re-running the completed job. +the durable session identity. Successful jobs may supply `result.message` with +`customType` (nonempty string), `content` (string), `display` (boolean), and optional +object `details`. The host publishes that existing custom-message payload without +plugin-specific rendering rules. Absent projections use the generic job outcome; +malformed projections keep the committed result and show a generic outcome with +an explicit projection diagnostic. Failed jobs always show the generic error. +Failed outcome persistence is visible without re-running the completed job. Supervisor entrypoint, worker entrypoint, `mcp_login::{WorkerMcpLoginUi, wire_worker_mcp_login}` (the worker's browser+callback login behind `mcp.begin_login`; wired by the agent engine before sessions register host handlers), client connection API for pa-tui/pa-cli, `acp::daemon::{run_daemon_attached_acp_mode, DaemonAcpOptions}` (pa-cli dispatches `--mode acp` through it), `agent_messaging::LinkAgentMessageController` + `rlm_children::{SupervisorChildSessions, ParentIdentity, RlmChildIdentity}` (e2e verifiers construct the worker-side family controller and the children registry; the engine wires the same types), `agent_engine::AgentSessionEngine::dispose_kernel` (the session-end kernel teardown the worker invokes at kill/shutdown/orphan exit — the engine outlives the session, so the pa-core engine-drop teardown cannot run there), the worker's `get_mcp_connections` command (`mcp_connections.rs`: the `/mcp` view's roster from the session's MCP manager overlaid with the session kernel's per-server tool listing, plus the api-key credential rows the same response serves — Rust-native session-plane extension; the TS daemon has no counterpart). Supervision internals `pub(crate)`. diff --git a/crates/pa-daemon/src/plugin_commands.rs b/crates/pa-daemon/src/plugin_commands.rs index 0a155f4ff..584fef663 100644 --- a/crates/pa-daemon/src/plugin_commands.rs +++ b/crates/pa-daemon/src/plugin_commands.rs @@ -689,18 +689,22 @@ return { #[tokio::test] async fn deferred_job_success_and_failure_reach_custom_message_transport_once() { - for fail in [false, true] { + for mode in ["generic", "failed", "projected", "malformed"] { let fixture = Fixture::new().await; std::fs::write( fixture.cwd.join("counter.lua"), - r#" + r" return {commands={{name='later',run=function(input,ctx) - return ctx.call('jobs.enqueue',{hook='later',input={fail=input.args=='fail'}}) + return ctx.call('jobs.enqueue',{hook='later',input={mode=input.args}}) end}},hooks={{name='later',run=function(input) - if input.fail then error('fixture deferred failure') end - return {summary='fixture deferred success'} + if input.mode=='failed' then error('fixture deferred failure') end + local result={summary='fixture deferred success'} + if input.mode=='projected' then + result.message={customType='fixture_outcome',content='plugin-owned presentation',display=true,details={source='fixture'}} + elseif input.mode=='malformed' then result.message={customType=42} end + return result end}}} -"#, +", ) .unwrap(); let reload = fixture.command("/reload", None).await; @@ -727,23 +731,19 @@ end}}} crate::plugin_job_events::publish(&core, &events, session, &job); })); let mut received = fixture.worker.events.subscribe(); - let command = fixture - .command( - if fail { - "/later fail" - } else { - "/later success" - }, - None, - ) - .await; + let command = fixture.command(&format!("/later {mode}"), None).await; assert!(command.success, "{command:?}"); let row = tokio::time::timeout(std::time::Duration::from_secs(5), async { loop { let frame = received.recv().await.unwrap(); let frame: Value = serde_json::from_slice(&frame.payload).unwrap(); if frame["event"]["type"] == "message_end" - && frame["event"]["message"]["customType"] == "plugin_job_outcome" + && frame["event"]["message"]["customType"] + == if mode == "projected" { + "fixture_outcome" + } else { + "plugin_job_outcome" + } { break frame["event"]["message"].clone(); } @@ -753,16 +753,37 @@ end}}} .expect("terminal outcome reaches attached client"); let terminal = fixture.core.drain_plugin_jobs().await; assert_eq!(terminal.len(), 1, "{terminal:?}"); - assert_eq!(row["details"], terminal[0]); + if mode == "projected" { + assert_eq!( + row, + json!({"role":"custom","customType":"fixture_outcome", + "content":"plugin-owned presentation","display":true,"details":{"source":"fixture"}}) + ); + } else if mode == "malformed" { + assert!(row["content"] + .as_str() + .unwrap() + .contains("invalid custom-message projection")); + let mut expected = terminal[0].clone(); + expected["projectionError"] = + json!("Plugin job returned an invalid custom-message projection"); + assert_eq!(row["details"], expected); + } else { + assert_eq!(row["details"], terminal[0]); + } assert_eq!(row["display"], true); assert_eq!( - row["details"]["state"]["status"], - if fail { "failed" } else { "completed" } + terminal[0]["state"]["status"], + if mode == "failed" { + "failed" + } else { + "completed" + } ); - assert!(row["content"].as_str().unwrap().contains(if fail { - "fixture deferred failure" - } else { - "fixture deferred success" + assert!(row["content"].as_str().unwrap().contains(match mode { + "failed" => "fixture deferred failure", + "projected" => "plugin-owned presentation", + _ => "fixture deferred success", })); assert_eq!( fixture.core.drain_plugin_jobs().await, @@ -773,7 +794,12 @@ end}}} fixture .rows() .iter() - .filter(|row| row["customType"] == "plugin_job_outcome") + .filter(|row| row["customType"] + == if mode == "projected" { + "fixture_outcome" + } else { + "plugin_job_outcome" + }) .count(), 1 ); diff --git a/crates/pa-daemon/src/plugin_job_events.rs b/crates/pa-daemon/src/plugin_job_events.rs index ead3071b6..650e9ba91 100644 --- a/crates/pa-daemon/src/plugin_job_events.rs +++ b/crates/pa-daemon/src/plugin_job_events.rs @@ -32,6 +32,27 @@ pub(crate) fn publish( }; let mut message = json!({"role":"custom","customType":"plugin_job_outcome", "content":content,"display":true,"details":job}); + if state["status"] == "completed" { + if let Some(projected) = state["result"].get("message") { + if projected + .get("customType") + .and_then(Value::as_str) + .is_some_and(|kind| !kind.is_empty()) + && projected.get("content").is_some_and(Value::is_string) + && projected.get("display").is_some_and(Value::is_boolean) + && projected.get("details").is_none_or(Value::is_object) + { + message = json!({"role":"custom","customType":projected["customType"], + "content":projected["content"],"display":projected["display"], + "details":projected.get("details")}); + } else { + let diagnostic = "Plugin job returned an invalid custom-message projection"; + eprintln!("pa-daemon: {diagnostic}"); + message["content"] = json!(format!("{content}\n{diagnostic}")); + message["details"]["projectionError"] = json!(diagnostic); + } + } + } if let Err(error) = crate::worker::persist_custom_row(&mut core, &message) { // The job already committed. Surface the persistence diagnostic without // re-running it or concealing the original terminal outcome. -- knotmirror 0.0.1