/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

//! Pure, disposable presentation of committed invocation facts. Model context is
//! a separate projection; retaining interrupted text here never accepts it there.
mod compact;
mod executor;
mod import;
mod message;
pub mod navigation;
pub mod shell;
mod step;
mod tools;
mod user;
use maka_agent_facts::event::Fact;
use maka_agent_facts::event::Invocation;
use maka_agent_facts::event::InvocationInput;
use maka_agent_facts::event::StoredEvent;
use maka_agent_facts::history::RowPosition;
use maka_agent_facts::tool_output::ToolOutput;
use message::timestamp;
pub use message::tool_message_id;
pub use message::{
    Content, Hidden, Message, Thinking, ToolContent, ToolMetadata, TurnState, Visible,
};

#[derive(Debug, thiserror::Error)]
pub enum ProjectionError {
    #[error("invalid presentation fact: {0}")]
    Invalid(&'static str),
    #[error("presentation text exceeds its byte limit")]
    TooLarge,
    #[error("presentation timestamp or sequence exceeds the wire range")]
    OutOfRange,
    #[error("unsupported presentation content: {0}")]
    Unsupported(&'static str),
}

#[derive(Debug, PartialEq)]
pub struct Row {
    /// Original emitting fact and its full row ordinal, before UI filtering.
    pub position: RowPosition,
    pub message: Message,
}

enum State {
    Vacant,
    Compact(Invocation),
    Active {
        invocation: Invocation,
        step: Option<Box<step::Step>>,
    },
    Ended,
    Poisoned,
}

pub struct InvocationView {
    executor: Option<executor::Execution>,
    state: State,
    last_sequence: u64,
    tools: tools::Tools,
    summary_step: Option<String>,
    summary_start: Option<compact::SummaryStart>,
}

impl Default for InvocationView {
    fn default() -> Self {
        Self {
            executor: None,
            state: State::Vacant,
            last_sequence: 0,
            tools: tools::Tools::new(),
            summary_step: None,
            summary_start: None,
        }
    }
}

impl InvocationView {
    pub fn new() -> Self {
        Self::default()
    }

    /// Input is one invocation's strictly ordered, committed facts, with gaps
    /// permitted for other sessions. Never feed the same fact twice.
    pub fn push(&mut self, stored: &StoredEvent) -> Result<Vec<Row>, ProjectionError> {
        self.push_with_tool_output(stored, None)
    }

    /// Successful T2 requires its caller-verified raw payload. Resolution and
    /// identity/digest verification belong to the durable store, not this view.
    pub fn push_with_tool_output(
        &mut self,
        stored: &StoredEvent,
        resolved: Option<&ToolOutput>,
    ) -> Result<Vec<Row>, ProjectionError> {
        let result = self.apply(stored, resolved);
        if result.is_err() {
            self.state = State::Poisoned;
        }
        result
    }

    fn apply(
        &mut self,
        stored: &StoredEvent,
        resolved: Option<&ToolOutput>,
    ) -> Result<Vec<Row>, ProjectionError> {
        let succeeds = matches!(
            stored.event.fact,
            Fact::ToolSettled {
                outcome: maka_agent_facts::event::ToolOutcome::Succeeded { .. },
                ..
            }
        );
        if succeeds != resolved.is_some() {
            return Err(ProjectionError::Invalid(
                "tool payload does not match fact kind",
            ));
        }
        if stored.sequence <= self.last_sequence {
            return Err(ProjectionError::Invalid("non-increasing log sequence"));
        }
        let event = &stored.event;
        let ts = timestamp(event)?;
        RowPosition {
            source_sequence: stored.sequence,
            item_index: 0,
        }
        .validate()
        .map_err(|_| ProjectionError::OutOfRange)?;
        if let Some(messages) = self.compact(event, ts)? {
            self.last_sequence = stored.sequence;
            return rows(stored.sequence, messages);
        }
        let mut messages = Vec::new();
        match &event.fact {
            Fact::MessageImported { record, .. } => {
                if !matches!(self.state, State::Vacant) {
                    return Err(ProjectionError::Invalid("import inside execution"));
                }
                messages.extend(import::project(event, record, ts)?);
                self.state = State::Ended;
            }
            Fact::InvocationOpened { input, .. } => {
                if !matches!(self.state, State::Vacant) {
                    return Err(ProjectionError::Invalid("duplicate invocation opening"));
                }
                if let InvocationInput::Message {
                    content,
                    source_messages,
                    ..
                } = input
                {
                    if source_messages
                        .iter()
                        .any(|source| source.message.origin.is_some())
                    {
                        for source in source_messages {
                            messages.push(user::project(
                                &source.message.message_id,
                                &event.invocation.turn_id,
                                ts,
                                &source.message.content,
                                source.message.origin.as_ref(),
                            )?);
                        }
                    } else {
                        messages.push(user::project(
                            if source_messages.len() == 1 {
                                &source_messages[0].message.message_id
                            } else {
                                &event.id
                            },
                            &event.invocation.turn_id,
                            ts,
                            content,
                            None,
                        )?);
                    }
                } else if input.inherited_claim().is_none() {
                    return Err(ProjectionError::Unsupported("Code invocation"));
                }
                messages.push(Message {
                    id: format!("{}:state", event.id),
                    turn_id: event.invocation.turn_id.clone(),
                    ts,
                    content: Content::TurnState {
                        state: TurnState::Running,
                    },
                });
                self.state = State::Active {
                    invocation: event.invocation.clone(),
                    step: None,
                };
            }
            fact => {
                let State::Active { invocation, step } = &mut self.state else {
                    return Err(ProjectionError::Invalid("fact outside active invocation"));
                };
                if *invocation != event.invocation {
                    return Err(ProjectionError::Invalid("invocation identity changed"));
                }
                match fact {
                    Fact::ExecutorStarted { binding, .. } => {
                        if step.is_some() || self.executor.is_some() {
                            return Err(ProjectionError::Invalid("overlapping execution backends"));
                        }
                        self.executor = Some(executor::Execution::new(event, ts, binding));
                    }
                    Fact::ExecutorObserved { output } => {
                        messages.extend(
                            self.executor
                                .as_mut()
                                .ok_or(ProjectionError::Invalid("executor output without start"))?
                                .observe(event, ts, output)?,
                        );
                    }
                    Fact::ExecutorCompleted { text } => {
                        messages.extend(
                            self.executor
                                .as_mut()
                                .ok_or(ProjectionError::Invalid(
                                    "executor completion without start",
                                ))?
                                .complete(event, ts, text)?,
                        );
                    }
                    Fact::MessageSteered { message, .. } => {
                        if step.is_some() {
                            return Err(ProjectionError::Invalid(
                                "steering inside a model request",
                            ));
                        }
                        messages.push(user::project(
                            &message.message_id,
                            &event.invocation.turn_id,
                            ts,
                            &message.content,
                            message.origin.as_ref(),
                        )?);
                    }
                    Fact::ModelRequested {
                        step_id,
                        model_id,
                        item_acceptance,
                        ..
                    } => {
                        if step.is_some() {
                            return Err(ProjectionError::Invalid("overlapping model steps"));
                        }
                        *step = Some(Box::new(step::Step::new(
                            step_id.clone(),
                            model_id.clone(),
                            *item_acceptance,
                        )));
                    }
                    Fact::ModelObserved {
                        step_id,
                        event: observation,
                    } => {
                        let step = step
                            .as_mut()
                            .ok_or(ProjectionError::Invalid("observation without request"))?;
                        step.check_id(step_id)?;
                        step.observe(event, ts, observation)?;
                        if step.item_acceptance {
                            match observation {
                                maka_agent_facts::model::ModelEvent::PartFinished {
                                    id, ..
                                } => {
                                    messages.extend(step.deliver_text(id));
                                }
                                maka_agent_facts::model::ModelEvent::Source(_)
                                | maka_agent_facts::model::ModelEvent::ToolCall(_)
                                | maka_agent_facts::model::ModelEvent::ProviderToolResult {
                                    ..
                                } => {
                                    if let Some((index, part)) = step.accepted_tool() {
                                        messages.extend(self.tools.accept_item(
                                            event,
                                            ts,
                                            step_id,
                                            index,
                                            part,
                                            step.anchor(),
                                            step.model(),
                                        )?);
                                    }
                                }
                                _ => {}
                            }
                        }
                    }
                    Fact::ModelCompleted { step_id, output } => {
                        let completed = step
                            .take()
                            .ok_or(ProjectionError::Invalid("completion without request"))?;
                        completed.check_id(step_id)?;
                        let streaming = completed.item_acceptance;
                        let text = completed.complete(output)?;
                        if streaming {
                            messages.extend(text);
                        } else {
                            messages.extend(self.tools.accept(event, ts, step_id, output, text)?);
                        }
                        if let (Some(input), Some(output_tokens)) =
                            (output.usage.input_tokens, output.usage.output_tokens)
                        {
                            messages.push(Message {
                                id: event.id.clone(),
                                turn_id: invocation.turn_id.clone(),
                                ts,
                                content: Content::TokenUsage {
                                    input,
                                    output: output_tokens,
                                    cache_read: output.usage.cache_read_tokens,
                                    cache_creation: output.usage.cache_write_tokens,
                                    reasoning: output.usage.reasoning_tokens,
                                },
                            });
                        }
                    }
                    Fact::ModelInterrupted {
                        step_id,
                        status,
                        diagnostic,
                        ..
                    } => {
                        let interrupted = step
                            .take()
                            .ok_or(ProjectionError::Invalid("interruption without request"))?;
                        interrupted.check_id(step_id)?;
                        messages.extend(interrupted.partial(
                            *status != maka_agent_facts::event::ModelInterruption::Cancelled,
                        ));
                        if let Some(message) = diagnostic
                            && *status != maka_agent_facts::event::ModelInterruption::Cancelled
                        {
                            messages.push(Message {
                                id: event.id.clone(),
                                turn_id: invocation.turn_id.clone(),
                                ts,
                                content: Content::ModelInterruption {
                                    message: message.clone(),
                                },
                            });
                        }
                    }
                    Fact::InvocationEnded { outcome } => {
                        if let Some(executor) = self.executor.take() {
                            messages.extend(executor.finish(event, ts));
                        }
                        if let Some(unfinished) = step.take() {
                            if matches!(
                                outcome,
                                maka_agent_facts::event::InvocationOutcome::Completed
                                    | maka_agent_facts::event::InvocationOutcome::HandoffPaused { .. }
                            ) {
                                return Err(ProjectionError::Invalid(
                                    "completion with unfinished model",
                                ));
                            }
                            // Failure seals observation evidence, not accepted
                            // model history. Never invent a completed response.
                            messages.extend(unfinished.partial(matches!(
                                outcome,
                                maka_agent_facts::event::InvocationOutcome::Failed { .. }
                            )));
                        }
                        if let Some(state) = TurnState::from_outcome(outcome, ts) {
                            messages.push(Message {
                                id: event.id.clone(),
                                turn_id: invocation.turn_id.clone(),
                                ts,
                                content: Content::TurnState { state },
                            });
                        }
                        self.state = State::Ended;
                    }
                    Fact::ToolDispatched { .. }
                    | Fact::ToolSettled { .. }
                    | Fact::ToolRejected { .. } => {
                        messages.extend(self.tools.boundary(event, ts, resolved)?);
                    }
                    Fact::InvocationOpened { .. } | Fact::MessageImported { .. } => unreachable!(),
                    Fact::ContextCheckpointRecorded { .. }
                    | Fact::ToolResultArchived { .. }
                    | Fact::ToolOperationObserved { .. }
                    | Fact::ToolNotified { .. } => {}
                }
            }
        }
        self.last_sequence = stored.sequence;
        rows(stored.sequence, messages)
    }

    /// Frozen caller-owned copies of observed text, including closed parts whose
    /// model completion has not committed. These are overlay, not durable rows.
    pub fn overlay(&self) -> Vec<Message> {
        if let Some(executor) = &self.executor {
            return executor.overlay();
        }
        match &self.state {
            State::Active {
                step: Some(step), ..
            } => step.partial(false),
            _ => Vec::new(),
        }
    }
}

fn rows(source_sequence: u64, messages: Vec<Message>) -> Result<Vec<Row>, ProjectionError> {
    messages
        .into_iter()
        .enumerate()
        .map(|(index, message)| {
            let position = RowPosition {
                source_sequence,
                item_index: u64::try_from(index).map_err(|_| ProjectionError::OutOfRange)?,
            };
            position
                .validate()
                .map_err(|_| ProjectionError::OutOfRange)?;
            Ok(Row { position, message })
        })
        .collect()
}