bamboo-engine 2026.10.22

//! Convert only a canonical, completed Child checkpoint to a Submission. This
//! is Host postprocessing, not a Worker-declared Artifact or quality verdict.
use super::*;
use bamboo_domain::{MessagePhase, Role};
use bamboo_tickets::OperationReceipt;

pub const TICKET_OWNED_STOP_KEY: &str = "ticket.runtime.owned_stop.v1";

/// A question ends the local one-shot Run without claiming a delivery. Only
/// the authoritative Assignment and exact raw Task operation receipt can
/// grant this broker-terminal exception; Worker metadata cannot grant it.
pub fn question_terminal_call_id(
    service: &TicketService,
    session: &Session,
    run_id: &str,
) -> Result<Option<String>> {
    let Some(dispatch) = read_dispatch(session)? else {
        return Ok(None);
    };
    let snapshot = service.published()?.1;
    let a = snapshot
        .assignments
        .get(&dispatch.assignment_id)
        .ok_or_else(|| Error::ScopeDenied("yield Assignment missing".into()))?;
    let Some(request_id) = &a.awaiting_request else {
        return Ok(None);
    };
    let receipt = a
        .runtime
        .as_ref()
        .ok_or_else(|| Error::ScopeDenied("yield admission missing".into()))?;
    if session.kind != SessionKind::Child
        || snapshot.binding != dispatch.binding
        || dispatch.receipt.as_ref() != Some(receipt)
        || receipt.run_id != run_id
        || receipt.session_id != session.id
        || dispatch.receipt_created_at != Some(session.created_at)
        || session.parent_session_id.as_deref()
            != Some(snapshot.binding.supervisor_session_id.as_str())
    {
        return Err(Error::ScopeDenied(
            "yield execution identity changed".into(),
        ));
    }
    let last = session
        .messages
        .last()
        .filter(|m| m.role == Role::Tool && m.tool_success == Some(true))
        .ok_or_else(|| Error::ScopeDenied("yield Tool result missing".into()))?;
    let call_id = last
        .tool_call_id
        .as_ref()
        .ok_or_else(|| Error::ScopeDenied("yield call missing".into()))?;
    let q = &snapshot.requests[request_id];
    let raw = session
        .metadata
        .get(crate::ticket_worker_plan::TICKET_QUESTION_YIELD_KEY)
        .ok_or_else(|| Error::ScopeDenied("Host yield question missing".into()))?;
    let original: bamboo_tickets::PendingRequest = serde_json::from_str(raw)?;
    if original.id != q.id
        || original.work_id != a.work_id
        || original.assignment_id.as_deref() != Some(a.id.as_str())
        || original.generation != a.generation
        || original.contract_revision != a.contract_revision
        || original.prompt_revision != q.prompt_revision
        || original.prompt != q.prompt
        || original.worker_packet_hash != q.worker_packet_hash
        || original.kind != bamboo_tickets::RequestKind::Question
        || serde_json::from_str::<serde_json::Value>(&last.content)?
            != serde_json::json!({"status":"waiting_for_answer", "request":original})
    {
        return Err(Error::ScopeDenied("yield question result changed".into()));
    }
    let calls: Vec<_> = session
        .messages
        .iter()
        .flat_map(|m| m.tool_calls.iter().flatten())
        .filter(|call| &call.id == call_id)
        .collect();
    if calls.len() != 1 || calls[0].function.name != "Task" {
        return Err(Error::ScopeDenied("yield Task pair missing".into()));
    }
    let authority = Authority::from_verified_host(
        snapshot.binding,
        Principal::Worker {
            assignment_id: a.id.clone(),
            generation: a.generation,
            run_id: run_id.into(),
            session_id: session.id.clone(),
        },
    );
    let source = bamboo_tickets::CommandSource::WorkerTask {
        arguments: serde_json::from_str(&calls[0].function.arguments)?,
    };
    if service
        .replay_source_command(
            &authority,
            &format!("worker-plan/{}/{call_id}", a.dispatch_key),
            &source,
        )?
        .is_none()
    {
        return Err(Error::ScopeDenied(
            "yield Task source receipt missing".into(),
        ));
    }
    Ok(Some(call_id.clone()))
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct OwnedStopCheckpoint {
    receipt: RuntimeReceipt,
    child_birth: chrono::DateTime<chrono::Utc>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pid: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    task_instance: Option<uuid::Uuid>,
    completed: bool,
}

/// Only a reaped local process from the exact current Host driver enters this
/// port. Persist its independent stop fact before Ticket publication, so a
/// failed Ticket write cannot turn a lease or display status into stop proof.
pub async fn checkpoint_owned_ticket_stop(
    service: &TicketService,
    runtime: &SessionInboxRuntimeBinding,
    session: &mut Session,
    run_id: &str,
    proof: &bamboo_subagent::fleet::ConfirmedLocalStop,
    completed: bool,
) -> Result<()> {
    checkpoint_owned_stop(
        service,
        runtime,
        session,
        run_id,
        Some(proof.pid()),
        None,
        completed,
    )
    .await
}

/// Same Ticket receipt contract, backed by an actually joined in-process task
/// and its drained resource domain. No synthetic PID is written.
pub async fn checkpoint_owned_task_stop(
    service: &TicketService,
    runtime: &SessionInboxRuntimeBinding,
    session: &mut Session,
    run_id: &str,
    proof: &crate::external_agents::local_task::ConfirmedTaskStop,
    completed: bool,
) -> Result<()> {
    if proof.session_id() != session.id
        || proof.child_birth() != session.created_at
        || proof.run_id() != run_id
    {
        return Err(Error::ScopeDenied(
            "task stop execution identity changed".into(),
        ));
    }
    checkpoint_owned_stop(
        service,
        runtime,
        session,
        run_id,
        None,
        Some(proof.instance_id()),
        completed,
    )
    .await
}

async fn checkpoint_owned_stop(
    service: &TicketService,
    runtime: &SessionInboxRuntimeBinding,
    session: &mut Session,
    run_id: &str,
    pid: Option<u32>,
    task_instance: Option<uuid::Uuid>,
    completed: bool,
) -> Result<()> {
    let dispatch = read_dispatch(session)?
        .ok_or_else(|| Error::ScopeDenied("stop dispatch missing".into()))?;
    let receipt = dispatch
        .receipt
        .ok_or_else(|| Error::ScopeDenied("stop Runtime receipt missing".into()))?;
    if receipt.session_id != session.id
        || receipt.run_id != run_id
        || dispatch.receipt_created_at != Some(session.created_at)
    {
        return Err(Error::ScopeDenied("stop execution identity changed".into()));
    }
    let checkpoint = OwnedStopCheckpoint {
        receipt: receipt.clone(),
        child_birth: session.created_at,
        pid,
        task_instance,
        completed,
    };
    let raw = serde_json::to_string(&checkpoint)?;
    let expected_birth = session.created_at;
    let expected_receipt = receipt.clone();
    let store = runtime.parent_question_lock.as_ref().ok_or_else(|| {
        Error::AuthorityUnavailable("canonical stop write coordinator unavailable".into())
    })?;
    let saved = store
        .mutate_runtime_session_and_publish(
            &session.id,
            || None,
            |current| {
                let exact = read_dispatch(current).ok().flatten().is_some_and(|d| {
                    d.receipt.as_ref() == Some(&expected_receipt)
                        && d.receipt_created_at == Some(expected_birth)
                });
                if current.created_at != expected_birth || !exact {
                    return Err(Error::ScopeDenied(
                        "stop Child birth or receipt changed".into(),
                    ));
                }
                if current
                    .metadata
                    .get(TICKET_OWNED_STOP_KEY)
                    .is_some_and(|previous| previous != &raw)
                {
                    return Err(Error::ScopeDenied("stop checkpoint already differs".into()));
                }
                if !current.metadata.contains_key(TICKET_OWNED_STOP_KEY) {
                    current
                        .metadata
                        .insert(TICKET_OWNED_STOP_KEY.into(), raw.clone());
                    current.metadata_version =
                        current.metadata_version.checked_add(1).ok_or_else(|| {
                            Error::AuthorityUnavailable("stop metadata revision overflow".into())
                        })?;
                }
                Ok::<_, Error>(())
            },
            |_| {},
        )
        .await??
        .ok_or_else(|| Error::AuthorityUnavailable("stop Child disappeared".into()))?;
    if saved.metadata.get(TICKET_OWNED_STOP_KEY) != Some(&raw) {
        return Err(Error::ScopeDenied("stop Child checkpoint rejected".into()));
    }
    session.metadata.insert(TICKET_OWNED_STOP_KEY.into(), raw);
    session.metadata_version = saved.metadata_version;
    checkpoint_stop_operation(
        service,
        &dispatch.binding,
        &dispatch.assignment_id,
        &checkpoint,
    )?;
    Ok(())
}

fn checkpoint_stop_operation(
    service: &TicketService,
    binding: &ScopeBinding,
    assignment_id: &str,
    checkpoint: &OwnedStopCheckpoint,
) -> Result<()> {
    let authority = Authority::from_verified_host(binding.clone(), Principal::Runtime);
    execute_runtime_command(
        service,
        &authority,
        &format!(
            "runtime-stop/{}/{}",
            checkpoint.receipt.dispatch_key, checkpoint.receipt.run_id
        ),
        vec![Operation::RuntimeStopped {
            assignment_id: assignment_id.into(),
            receipt: checkpoint.receipt.clone(),
            completed: checkpoint.completed,
        }],
    )?;
    Ok(())
}

pub async fn checkpoint_ticket_result(
    service: &TicketService,
    storage: &dyn Storage,
    expected: &Session,
    run_id: &str,
) -> Result<Option<OperationReceipt>> {
    if !expected
        .metadata
        .contains_key(crate::ticket_worker_plan::TICKET_LOCAL_PLAN_KEY)
    {
        return Ok(None);
    }
    let canonical = storage
        .load_session(&expected.id)
        .await?
        .ok_or_else(|| Error::AuthorityUnavailable("canonical Child checkpoint missing".into()))?;
    let dispatch = read_dispatch(&canonical)?
        .ok_or_else(|| Error::ScopeDenied("canonical dispatch receipt missing".into()))?;
    let receipt = dispatch
        .receipt
        .as_ref()
        .ok_or_else(|| Error::ScopeDenied("Runtime receipt missing".into()))?;
    if canonical.kind != SessionKind::Child
        || canonical.created_at != expected.created_at
        || canonical.parent_session_id != expected.parent_session_id
        || canonical.root_session_id != expected.root_session_id
        || dispatch.receipt_created_at != Some(canonical.created_at)
        || receipt.run_id != run_id
        || receipt.session_id != canonical.id
        || receipt.dispatch_key != dispatch.dispatch_key
        || receipt.spec_hash != dispatch.spec_hash
    {
        return Err(Error::ScopeDenied(
            "completed Child execution identity changed".into(),
        ));
    }
    let terminal_status = canonical.last_run_status();
    if !matches!(
        terminal_status.as_deref(),
        Some("completed" | "error" | "timeout" | "cancelled" | "skipped")
    ) {
        return Ok(None);
    }
    let (binding, _) = verified_scope_binding_for_service(
        storage,
        service,
        &dispatch.binding.supervisor_session_id,
    )
    .await?;
    let snapshot = service.published()?.1;
    let assignment = snapshot
        .assignments
        .get(&dispatch.assignment_id)
        .ok_or_else(|| Error::ScopeDenied("Assignment missing".into()))?;
    if snapshot.binding != binding
        || dispatch.binding != binding
        || snapshot
            .intents
            .get(&dispatch.dispatch_key)
            .is_none_or(|i| i.spec_hash != dispatch.spec_hash)
        || assignment.runtime.as_ref() != Some(receipt)
    {
        return Err(Error::ScopeDenied(
            "completed Child differs from immutable Assignment".into(),
        ));
    }
    // Only Host-owned reaping writes this reserved canonical checkpoint. Its
    // recovery is required for failed/cancelled runs too, before any ACK.
    let checkpoint = canonical
        .metadata
        .get(TICKET_OWNED_STOP_KEY)
        .map(|raw| -> Result<OwnedStopCheckpoint> {
            if raw.len() > 4096 {
                return Err(Error::ScopeDenied("stop checkpoint exceeds limit".into()));
            }
            let checkpoint: OwnedStopCheckpoint = serde_json::from_str(raw)?;
            if checkpoint.receipt != *receipt
                || checkpoint.child_birth != canonical.created_at
                || !matches!(
                    (checkpoint.pid, checkpoint.task_instance),
                    (Some(1..), None) | (None, Some(_))
                )
                || checkpoint.task_instance.is_some_and(|instance| {
                    instance.is_nil()
                        || canonical
                            .metadata
                            .get("runtime.local_task.state")
                            .map(String::as_str)
                            != Some("stopped")
                        || canonical.metadata.get("runtime.local_task.instance")
                            != Some(&instance.to_string())
                })
                || terminal_status.as_deref() == Some("completed")
                    && !checkpoint.completed
                    && assignment.awaiting_request.is_none()
            {
                return Err(Error::ScopeDenied(
                    "canonical stop checkpoint mismatch".into(),
                ));
            }
            checkpoint_stop_operation(service, &snapshot.binding, &assignment.id, &checkpoint)?;
            Ok(checkpoint)
        })
        .transpose()?;
    if checkpoint.is_none() {
        return Err(Error::AuthorityUnavailable(
            "canonical owned-stop checkpoint missing; retain broker receipt".into(),
        ));
    }
    if terminal_status.as_deref() != Some("completed") {
        return Ok(None);
    }
    if let Some(request_id) = &assignment.awaiting_request {
        let question = snapshot
            .requests
            .get(request_id)
            .ok_or_else(|| Error::ScopeDenied("yield question missing".into()))?;
        if question.assignment_id.as_deref() != Some(assignment.id.as_str())
            || question.work_id != assignment.work_id
            || question.generation != assignment.generation
            || question.contract_revision != assignment.contract_revision
        {
            return Err(Error::ScopeDenied("yield question binding changed".into()));
        }
        // Ending a one-shot question run is a capacity release, not a delivery.
        // Its durable request/stop survives replay without a fake Submission.
        return Ok(None);
    }
    let output = canonical
        .messages
        .last()
        .filter(|m| {
            m.role == Role::Assistant
                && m.phase != Some(MessagePhase::Commentary)
                && m.tool_calls.as_ref().is_none_or(Vec::is_empty)
                && !m.content.trim().is_empty()
        })
        .ok_or_else(|| {
            Error::InvalidTransition("completed Child has no final output checkpoint".into())
        })?;
    let runtime = Authority::from_verified_host(binding.clone(), Principal::Runtime);
    // A prior Host saved this fact only after reaping its owned process.
    // Recover Ticket publication before any broker ACK, without launching a
    // run or granting Worker tools. Missing proof keeps old attempts fenced.
    if checkpoint.is_some() {
        let current = service.published()?.1;
        let a = &current.assignments[&assignment.id];
        let work = &current.tickets[&a.work_id];
        let submit_id = format!("runtime-submit/{}/{run_id}", dispatch.dispatch_key);
        if !current.receipts.contains_key(&submit_id)
            && a.process_stopped
            && a.generation == work.generation
            && a.contract_revision == work.contract_revision
            && work.active_assignment.as_deref() == Some(&a.id)
            && work.state != WorkState::Cancelled
            && (a.authority_epoch != current.authority_epoch
                || a.state == AssignmentState::OutcomeUnknown)
        {
            // Explicit canonical completion reconciliation. It never resumes
            // execution; process_stopped continues to deny tools/dispatch.
            let reconciled = execute_runtime_command(
                service,
                &runtime,
                &format!(
                    "runtime-completed-reconcile/{}/{}",
                    dispatch.dispatch_key, current.authority_epoch
                ),
                vec![Operation::ReconcileCompleted {
                    assignment_id: a.id.clone(),
                    receipt: receipt.clone(),
                }],
            );
            if let Err(error) = reconciled {
                if !matches!(
                    error,
                    Error::ResourceBlocked(_) | Error::InvalidTransition(_)
                ) {
                    return Err(error);
                }
                // Retain a stale Submission for changed inputs/unknown effects.
            }
        }
    }
    let artifact = service.store_artifact(&runtime, output.content.as_bytes())?;
    let mut artifacts = vec![artifact.clone()];
    for effect in service.published()?.1.assignments[&assignment.id]
        .effects
        .values()
    {
        if effect.state == bamboo_tickets::EffectState::Succeeded {
            if let Some(file) = &effect.artifact {
                if !artifacts.contains(file) {
                    artifacts.push(file.clone());
                }
            }
        }
    }
    let worker = Authority::from_verified_host(
        binding,
        Principal::Worker {
            assignment_id: assignment.id.clone(),
            generation: assignment.generation,
            run_id: receipt.run_id.clone(),
            session_id: receipt.session_id.clone(),
        },
    );
    let id = format!("runtime-submit/{}/{run_id}", dispatch.dispatch_key);
    let receipt = execute_runtime_command(
        service,
        &worker,
        &id,
        vec![Operation::Submit {
            assignment_id: assignment.id.clone(),
            temp_id: "submission".into(),
            evidence: vec![format!(
                "canonical Child {}; run {}; message {}; sha256 {}",
                canonical.id, run_id, output.id, artifact.sha256
            )],
            artifacts,
        }],
    )?;
    // Submit's generation/contract/input fence preserves a late result as
    // stale and cannot replace a newer generation or reverse cancellation.
    Ok(Some(receipt))
}