polyc_agent/

delegate.rs

//! Delegation (in-process sub-agent task) primitive for the agent turn loop.
//!
//! # Why a reserved tool name, joining the batch (not short-circuiting it)
//!
//! This mirrors [`crate::handoff`]'s reserved-tool-name design (a model-facing
//! function the loop recognizes by name, so no new provider integration or
//! wire channel is needed) but the two primitives have OPPOSITE control flow:
//!
//!   * A **handoff** suspends the whole turn — the parent conversation stops,
//!     a child `Conversation` is created, and the parent resumes only once
//!     the child's `HandoffReturn` lands (possibly turns later). It short-
//!     circuits the batch: no other tool in the same batch executes.
//!   * A **delegation** (`__delegate_to`, #870) runs a nested, context-
//!     isolated turn IN-PROCESS, synchronously, as part of dispatching this
//!     SAME batch — it joins `run_turn_with`'s ordinary `tool_futures`
//!     alongside every other call in the batch, and its result is just
//!     another `tool_result` the SAME turn's next provider step sees. There
//!     is no suspend, no child resource, no later turn.
//!
//! This is the tracer bullet for PRD #867: exactly one task, to exactly one
//! worker, capped at one level deep (a worker's own advertised tool set never
//! includes `__delegate_to` — see [`crate::run_turn_with`]'s tool-spec
//! pinning).

use std::sync::Arc;

use polyc_llm::{DynProvider, ToolSpec};

/// The reserved tool name the model emits to request an in-process
/// delegation to a scoped worker agent.
///
/// Advertised only when [`crate::RunTurnOptions::delegate_descriptors`] is
/// non-empty (see [`delegate_tool_spec`]'s call site in `run_turn_with`) — a
/// conversation whose agent declares no delegation targets never sees this
/// name at all, so it can't collide with a real tool of the same name either.
pub const DELEGATE_TOOL_NAME: &str = "__delegate_to";

/// The condensation contract appended to every delegated worker's
/// synthesized instructions when the call carries no `result_schema`
/// (INV-C25, #1140).
///
/// The worker's final message is the sole return channel back to the
/// caller, so the worker is told to make that message a self-contained
/// summary of the outcome. When a `result_schema` IS in force, the
/// schema-forced finalize path bounds the answer's shape instead and this
/// text is not injected. The per-call result cap (`MAX_TOOL_RESULT_BYTES`
/// middle-elision) stays as the hard backstop either way — this contract
/// instructs, the cap enforces.
pub const WORKER_CONDENSATION_CONTRACT: &str = "You are completing one delegated task. The \
    caller sees only your final message — none of your tool calls, intermediate work, or \
    earlier drafts reach it. Make your final message a self-contained summary of the outcome: \
    what you did or found, the key details the caller needs, and anything that failed. Keep it \
    concise — an overlong answer is trimmed from the middle.";

/// Composes a delegated worker's synthesized system message text
/// (INV-C25, `#1140`), given the descriptor's own (already trimmed,
/// non-empty-or-`None`) `instructions` and whether the call carries a
/// `result_schema`.
///
/// With no `result_schema` in force, the worker's final message is the sole
/// return channel, so [`WORKER_CONDENSATION_CONTRACT`] is always appended
/// (on its own line pair after `instructions`, or standalone when
/// `instructions` is `None`) — this branch never returns `None`. With a
/// `result_schema` in force, the schema-forced finalize path bounds the
/// answer's shape instead, so the contract text is NOT injected and
/// `instructions` passes through unchanged (`None` stays `None`).
#[must_use]
pub(crate) fn worker_system_text(
    instructions: Option<&str>,
    has_result_schema: bool,
) -> Option<String> {
    if has_result_schema {
        return instructions.map(str::to_owned);
    }
    Some(instructions.map_or_else(
        || WORKER_CONDENSATION_CONTRACT.to_owned(),
        |instructions| format!("{instructions}\n\n{WORKER_CONDENSATION_CONTRACT}"),
    ))
}

/// Renders a delegated worker's turn-start system message (`#1323`),
/// mirroring `polyc_control_plane`'s top-level `turn_start_block` — same
/// wording, same UTC-at-minute-precision rendering — from `unix_ms`, the
/// PARENT turn's frozen dispatch clock
/// (`RunTurnOptions::turn_start_unix_ms`), never an independent read: a
/// worker's nested turn has no dispatch clock of its own to freeze, and
/// reading one here would break replay determinism (INV-11).
///
/// `None` when `unix_ms` falls outside `jiff::Timestamp`'s representable
/// range (in practice, only a caller passing `u64::MAX`) — a worker told
/// nothing is safer than one told a wrong time, the same rule the top-level
/// stamp follows.
///
/// Pushed as its OWN system message (see [`crate::run_delegate_call`]),
/// never folded into [`worker_system_text`]'s returned string: the
/// instructions/condensation text is the worker prompt's stable content,
/// and this value changes on every dispatch, so joining them would defeat
/// any future caching of the stable part.
#[must_use]
pub(crate) fn worker_turn_start_block(unix_ms: u64) -> Option<String> {
    let instant = i64::try_from(unix_ms)
        .ok()
        .and_then(|ms| jiff::Timestamp::from_millisecond(ms).ok())?
        .strftime("%Y-%m-%d %H:%M")
        .to_string();
    Some(format!(
        "This turn started at {instant} UTC. Later steps in this turn may \
         run after this instant."
    ))
}

/// JSON-schema spec for the delegate tool. Provided alongside the user's tool
/// specs, but ONLY when at least one [`DelegateDescriptor`] is configured —
/// see [`delegate_tool_spec`].
#[must_use]
pub fn delegate_tool_spec() -> ToolSpec {
    // Like the handoff primitive, delegation is a runtime mechanism the
    // capability gate never mediates (the orchestrator-level call is always
    // allowed) — the worker's OWN nested turn re-applies the full gate to
    // everything it does, fail-closed (see `run_turn_with`'s unattended-mode
    // wiring for the nested options).
    ToolSpec::new(
        DELEGATE_TOOL_NAME,
        "Hand a single, self-contained task to a specialized worker and wait for its answer. \
         The worker runs in an isolated context — it does NOT see this conversation's history, \
         only `task` and, if given, `context` — so state everything the worker needs to know. \
         `target_agent_id` selects which worker runs the task. Set `result_schema` (a JSON \
         Schema) to force the worker's answer into that shape instead of free text — the worker \
         gets one retry if its first answer doesn't match, and reports a structured failure if it \
         still can't conform.",
        serde_json::json!({
            "type": "object",
            "properties": {
                "target_agent_id": {
                    "type": "string",
                    "description": "Identifier of the worker agent to run the task."
                },
                "task": {
                    "type": "string",
                    "description": "The self-contained task for the worker to perform."
                },
                "context": {
                    "type": "string",
                    "description": "Optional extra context the worker needs — the worker sees no \
                        other history, so include anything relevant here."
                },
                "result_schema": {
                    "type": "object",
                    "description": "Optional JSON Schema the worker's final answer must satisfy. \
                        Omit for a free-text answer."
                }
            },
            "required": ["target_agent_id", "task"],
            "additionalProperties": false
        }),
    )
}

/// A resolved, self-contained worker configuration for one `can_delegate_to`
/// target (#870).
///
/// Built by the control plane at turn dispatch — NEVER by this crate — and
/// threaded down through the wire (`TurnInput.delegate_descriptors`) and the
/// harness's tool-executor composition
/// (`polyc_turn_runner::resolve_delegate_descriptors`) into
/// [`crate::RunTurnOptions::delegate_descriptors`]. See
/// `crates/control-plane/src/delegate.rs` for how the fields here are
/// resolved (provider/model fallback, connector-scope intersection, the
/// read-only-by-default built-in allowlist).
#[derive(Clone)]
pub struct DelegateDescriptor {
    /// The target `Agent` resource name — matched (trailing-name, mirroring
    /// [`crate::HandoffRequest::child_agent_id`]'s resolution) against the
    /// model's `__delegate_to(target_agent_id, ...)` argument to pick this
    /// descriptor. See [`find_descriptor`].
    pub agent_id: String,
    /// System instructions for the worker's nested turn. `None` ⇒ no
    /// agent-specific instructions.
    pub instructions: Option<String>,
    /// The worker's resolved backend, already picked from the deployment's
    /// registered providers — this crate never resolves a provider selector
    /// string itself.
    pub provider: Arc<DynProvider>,
    /// The registry key of [`Self::provider`] (e.g. `"vertex"`, `"stub"`) —
    /// carried alongside the erased backend so a forensic record (`#872`,
    /// `DelegateRecord::resolved_provider`) can name the provider without
    /// this crate needing a `Debug`/name accessor on [`DynProvider`] itself.
    pub provider_name: String,
    /// The worker's resolved model id.
    pub model: String,
    /// The worker's advertised tool specs. Never includes
    /// [`DELEGATE_TOOL_NAME`] — this is what caps delegation depth at one,
    /// since [`crate::run_turn_with`] only advertises the delegate tool when
    /// its OWN `delegate_descriptors` option is non-empty, and a nested turn
    /// always runs with that option empty.
    pub tool_specs: Vec<ToolSpec>,
    /// The worker's step budget, applied to the nested turn's
    /// `RunTurnOptions::max_steps`.
    pub max_steps: usize,
    /// Whether the worker's nested turn may ground on the provider's native
    /// web-search primitive (`RunTurnOptions::native_search_allowed`).
    /// Derived the same way the parent turn's own scoping is (`#1226`): the
    /// resolved descriptor's `builtin_tools` named
    /// `polyc_tools::web::NATIVE_SEARCH_GROUNDING` — never hardcoded true,
    /// since that would hand every worker a capability its own agent
    /// manifest never granted.
    pub native_search_allowed: bool,
}

impl std::fmt::Debug for DelegateDescriptor {
    /// Hand-rolled: [`DynProvider`] carries no `Debug` impl (the `LlmProvider`
    /// trait doesn't require one), so this can't be `#[derive(Debug)]`d.
    /// Prints tool names, not full specs, to stay short in a turn-level log.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DelegateDescriptor")
            .field("agent_id", &self.agent_id)
            .field("provider_name", &self.provider_name)
            .field("model", &self.model)
            .field(
                "tool_specs",
                &self.tool_specs.iter().map(|s| &s.name).collect::<Vec<_>>(),
            )
            .field("max_steps", &self.max_steps)
            .finish_non_exhaustive()
    }
}

/// The trailing name segment of a `target_agent_id` / descriptor `agent_id`,
/// mirroring [`crate::handoff`]'s equivalent (own copy — see that module for
/// why the tolerant match exists: an operator may author either a bare name
/// or a namespaced `agent:ns/name` ref).
fn trailing_name(entry: &str) -> &str {
    entry.rsplit('/').next().unwrap_or(entry)
}

/// Find the [`DelegateDescriptor`] matching `target_agent_id` by trailing
/// name.
#[must_use]
pub fn find_descriptor<'a>(
    descriptors: &'a [DelegateDescriptor],
    target_agent_id: &str,
) -> Option<&'a DelegateDescriptor> {
    let target = trailing_name(target_agent_id);
    descriptors
        .iter()
        .find(|d| trailing_name(&d.agent_id) == target)
}

/// Parsed `__delegate_to` arguments, produced by [`parse_delegate_args`].
#[derive(Debug, Clone)]
pub struct DelegateRequest {
    /// Tool-call id the provider assigned to the `__delegate_to` call. Echoed
    /// back as the `tool_result` id so the function-calling loop sees a
    /// matched call → result pair.
    pub call_id: String,
    /// The model's chosen worker agent identifier.
    pub target_agent_id: String,
    /// The self-contained task for the worker to perform — becomes the sole
    /// user message of the worker's fresh transcript.
    pub task: String,
    /// Optional extra context, appended to the worker's transcript alongside
    /// `task`. `None` when the model supplied none.
    pub context: Option<String>,
    /// Optional JSON Schema the worker's final answer must satisfy (`#871`).
    /// `None` ⇒ the worker answers in free text under the
    /// [`WORKER_CONDENSATION_CONTRACT`] appended to its instructions
    /// (INV-C25, `#1140`). The raw schema value is not validated for well-formedness
    /// here (compiling it into a [`jsonschema::Validator`] is the caller's
    /// job, at the point it's actually used) — a bad schema is an argument
    /// error the caller surfaces the same way a missing `task` is.
    pub result_schema: Option<serde_json::Value>,
}

/// Parse the JSON arguments of a `__delegate_to` tool call into a structured
/// [`DelegateRequest`].
///
/// Returns `None` if `args_json` doesn't parse, or either required field
/// (`target_agent_id`, `task`) is missing or empty — the caller then
/// surfaces a legible tool-result error rather than dispatching a malformed
/// delegation.
#[must_use]
pub fn parse_delegate_args(call_id: &str, args_json: &str) -> Option<DelegateRequest> {
    let v: serde_json::Value = serde_json::from_str(args_json).ok()?;
    let target_agent_id = v.get("target_agent_id")?.as_str()?.to_owned();
    if target_agent_id.is_empty() {
        return None;
    }
    let task = v.get("task")?.as_str()?.to_owned();
    if task.is_empty() {
        return None;
    }
    let context = v
        .get("context")
        .and_then(serde_json::Value::as_str)
        .filter(|s| !s.is_empty())
        .map(str::to_owned);
    let result_schema = v.get("result_schema").cloned();
    Some(DelegateRequest {
        call_id: call_id.to_owned(),
        target_agent_id,
        task,
        context,
        result_schema,
    })
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use super::*;

    fn descriptor(agent_id: &str) -> DelegateDescriptor {
        DelegateDescriptor {
            agent_id: agent_id.to_owned(),
            instructions: None,
            provider: polyc_llm::into_dyn(polyc_llm::turn::StubProvider),
            provider_name: "stub".to_owned(),
            model: "stub".to_owned(),
            tool_specs: Vec::new(),
            max_steps: 4,
            native_search_allowed: false,
        }
    }

    #[test]
    fn parses_minimum_required_args() {
        let req = parse_delegate_args(
            "c-1",
            r#"{"target_agent_id":"researcher","task":"find the answer"}"#,
        )
        .unwrap();
        assert_eq!(req.target_agent_id, "researcher");
        assert_eq!(req.task, "find the answer");
        assert!(req.context.is_none());
        assert_eq!(req.call_id, "c-1");
    }

    #[test]
    fn parses_optional_context() {
        let req = parse_delegate_args(
            "c-2",
            r#"{"target_agent_id":"x","task":"t","context":"extra"}"#,
        )
        .unwrap();
        assert_eq!(req.context.as_deref(), Some("extra"));
    }

    #[test]
    fn parses_optional_result_schema() {
        let req = parse_delegate_args(
            "c-3",
            r#"{"target_agent_id":"x","task":"t","result_schema":{"type":"object"}}"#,
        )
        .unwrap();
        assert_eq!(
            req.result_schema,
            Some(serde_json::json!({"type":"object"}))
        );
    }

    #[test]
    fn result_schema_absent_by_default() {
        let req = parse_delegate_args("c-4", r#"{"target_agent_id":"x","task":"t"}"#).unwrap();
        assert!(req.result_schema.is_none());
    }

    #[test]
    fn rejects_missing_target_agent_id() {
        assert!(parse_delegate_args("c", r#"{"task":"t"}"#).is_none());
    }

    #[test]
    fn rejects_empty_target_agent_id() {
        assert!(parse_delegate_args("c", r#"{"target_agent_id":"","task":"t"}"#).is_none());
    }

    #[test]
    fn rejects_missing_task() {
        assert!(parse_delegate_args("c", r#"{"target_agent_id":"x"}"#).is_none());
    }

    #[test]
    fn rejects_empty_task() {
        assert!(parse_delegate_args("c", r#"{"target_agent_id":"x","task":""}"#).is_none());
    }

    #[test]
    fn rejects_garbage_json() {
        assert!(parse_delegate_args("c", "not-json").is_none());
    }

    #[test]
    fn delegate_tool_spec_has_required_fields() {
        let spec = delegate_tool_spec();
        assert_eq!(spec.name, DELEGATE_TOOL_NAME);
        let required = spec
            .schema_json
            .get("required")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        assert!(required.iter().any(|v| v == "target_agent_id"));
        assert!(required.iter().any(|v| v == "task"));
        // #1141: strict schema — no undeclared arguments.
        assert_eq!(
            spec.schema_json.get("additionalProperties"),
            Some(&serde_json::json!(false))
        );
    }

    #[test]
    fn find_descriptor_matches_by_trailing_name() {
        let descriptors = vec![descriptor("agent:default/researcher"), descriptor("coder")];
        assert!(find_descriptor(&descriptors, "researcher").is_some());
        assert!(find_descriptor(&descriptors, "agent:other-ns/researcher").is_some());
        assert!(find_descriptor(&descriptors, "coder").is_some());
        assert!(find_descriptor(&descriptors, "ghost").is_none());
    }

    // ── #1140 / INV-C25: `worker_system_text` ────────────────────────────────
    //
    // TEST-17 (CONF-17), as direct unit tests of the pure composition helper
    // (PR #1152 review finding) rather than round-tripping a full worker turn
    // through a provider-capture harness — the schema×instructions matrix
    // lives entirely in this one function.

    /// TEST-17, first half: no `result_schema` ⇒ the contract is appended
    /// after the descriptor's own instructions.
    #[test]
    fn contract_appended_after_instructions_without_schema() {
        let text = worker_system_text(Some("You are a scoped worker."), false)
            .expect("no-schema path always returns Some");
        assert_eq!(
            text,
            format!("You are a scoped worker.\n\n{WORKER_CONDENSATION_CONTRACT}")
        );
    }

    /// TEST-17 corollary: no instructions of its own and no `result_schema`
    /// ⇒ the contract alone — a worker is never dispatched untold that its
    /// final message is the sole return channel.
    #[test]
    fn contract_alone_without_instructions_or_schema() {
        let text = worker_system_text(None, false).expect("no-schema path always returns Some");
        assert_eq!(text, WORKER_CONDENSATION_CONTRACT);
    }

    /// TEST-17, second half: with a `result_schema` in force, the
    /// schema-forced finalize path satisfies INV-C25 instead — the contract
    /// text is NOT injected and the descriptor's own instructions pass
    /// through unchanged.
    #[test]
    fn instructions_unchanged_with_schema() {
        let text = worker_system_text(Some("You are a scoped worker."), true);
        assert_eq!(text.as_deref(), Some("You are a scoped worker."));
    }

    /// With a `result_schema` in force AND no instructions, there is nothing
    /// to inject or pass through — no system message at all.
    #[test]
    fn no_system_text_with_schema_and_no_instructions() {
        assert_eq!(worker_system_text(None, true), None);
    }

    // ── #1323: `worker_turn_start_block` ────────────────────────────────────

    // Keep in lockstep with
    // `turn_start_block_labels_the_start_time_states_utc_and_omits_on_derivation_failure`
    // (crates/control-plane/src/grpc/tests.rs) — that test pins the same full
    // string for the same input ms against `turn_start_block`, the top-level
    // renderer this one deliberately mirrors.
    #[test]
    fn renders_utc_at_minute_precision() {
        // 2024-05-17T09:33:59Z, truncated to its own minute (never rounded).
        let block = worker_turn_start_block(1_715_938_439_000).expect("in-range");
        assert_eq!(
            block,
            "This turn started at 2024-05-17 09:33 UTC. Later steps in this turn may run after \
             this instant."
        );
    }

    #[test]
    fn matches_the_top_level_blocks_wording() {
        // Mirrors `polyc_control_plane::grpc::turn_start_block`'s phrasing
        // exactly — "started" framing, never "now".
        let block = worker_turn_start_block(0).expect("epoch is in range");
        assert!(block.starts_with("This turn started at "));
        assert!(!block.to_lowercase().contains("now"));
    }

    #[test]
    fn out_of_range_instant_renders_no_block() {
        assert_eq!(worker_turn_start_block(u64::MAX), None);
    }

    #[test]
    fn same_input_ms_renders_identical_bytes() {
        // Determinism: replaying the same recorded dispatch clock must
        // reproduce the exact same stamp, never a fresh clock's drift.
        let a = worker_turn_start_block(1_715_938_439_000);
        let b = worker_turn_start_block(1_715_938_439_000);
        assert_eq!(a, b);
    }
}