///|
/// `PortRuntime` — the default `Runtime`: delegates every effect to the
/// composed extension ports (`ModelPort` + `ToolProvider`s). Cancellation
/// routes to providers that registered a `ToolCancellable` facet for the
/// effect; every other id reports `NotPropagated`.
///
/// Host builders wrap this type and override the methods they need instead
/// of implementing `Runtime` from scratch.

///|
/// Map `@error.ModelError` (raised by ModelPort) to the kernel's
/// `ModelFailure` so it can travel through the reducer as data.
fn model_error_to_failure(error : @error.ModelError) -> @kernel.ModelFailure {
  match error {
    RequestBuild(reason) => RequestBuild(reason~)
    Transport(reason) => Transport(reason~)
    ResponseParse(reason) => Parse(reason~)
    RateLimited(info) =>
      RateLimited(
        reason=info.message,
        reset_at_ms=info.reset_at_ms,
        provider_code=info.provider_code,
      )
  }
}

///|
/// Decode the provider-agnostic subset of the canonical call-options bag.
///
/// `ChatOptions` intentionally cannot represent provider-specific fields. Do
/// not drop those fields: reject them at this boundary so the caller can route
/// the original bag to a host that understands it.
fn decode_chat_options(
  call_options : Json,
) -> Result[@types.ChatOptions, @kernel.ModelFailure] {
  let fields = match call_options {
    Null => return Ok({ temperature: None, max_output_tokens: None, })
    Object(fields) => fields
    _ =>
      return Err(RequestBuild(reason="call_options must be null or an object"))
  }
  for key in fields.keys() {
    if key != "temperature" && key != "max_output_tokens" {
      return Err(
        RequestBuild(reason="unsupported call_options field '" + key + "'"),
      )
    }
  }
  let temperature = match fields.get("temperature") {
    None => None
    Some(Number(value, ..)) =>
      if (value - value).is_nan() {
        return Err(
          RequestBuild(
            reason="call_options.temperature must be a finite number",
          ),
        )
      } else {
        Some(value)
      }
    Some(_) =>
      return Err(
        RequestBuild(reason="call_options.temperature must be a number"),
      )
  }
  let max_output_tokens = match fields.get("max_output_tokens") {
    None => None
    Some(Number(value, ..)) => {
      let converted = value.to_int()
      if converted.to_double() != value {
        return Err(
          RequestBuild(
            reason="call_options.max_output_tokens must be an integer in Int range",
          ),
        )
      }
      Some(converted)
    }
    Some(_) =>
      return Err(
        RequestBuild(reason="call_options.max_output_tokens must be a number"),
      )
  }
  Ok({ temperature, max_output_tokens, })
}

///|
priv struct ToolCancellationRegistration {
  mut handler : &@port.ToolCancellable?
  mut active : Bool
}

///|
pub struct PortRuntime {
  priv model : &@port.ModelPort
  /// Retain unsettled executions by run and effect. Legacy cancellation
  /// requests contain only effect ids, so ambiguous requests must fail closed.
  priv cancels : Map[(String, Int32), ToolCancellationRegistration]
}

///|
pub fn PortRuntime::PortRuntime(model~ : &@port.ModelPort) -> PortRuntime {
  { model, cancels: Map::from_array([]), }
}

///|
pub extend PortRuntime with Runtime::{
  cancel_effects,
  execute_tool,
  call_model,
  compact,
}

///|
pub impl Runtime for PortRuntime with fn call_model(
  self,
  scope : @kernel.InvocationScope,
  messages : ArrayView[@kernel.Message],
  tool_definitions : Array[@kernel.ToolDef],
  call_options : Json,
  on_chunk : ((@types.StreamChunk) -> Unit)?,
) -> Result[@kernel.ModelCallResult, @kernel.ModelFailure] {
  let options = match decode_chat_options(call_options) {
    Ok(options) => options
    Err(failure) => return Err(failure)
  }
  let stream : @types.StreamMode = match on_chunk {
    Some(cb) => Stream(cb)
    None => NoStream
  }
  let result = self.model.chat(
    scope, messages, tool_definitions, options, stream,
  ) catch {
    error => return Err(model_error_to_failure(error))
  }
  // When the modelport returns no preprocessing (empty processed_messages),
  // substitute the input messages so the reducer's transcript-replace step
  // is a no-op. Real modelports that actually preprocess return their
  // processed tree here and we honour it. The view must be copied before being
  // stored/returned — this is the single retained O(N) copy per call.
  let final_result : @kernel.ModelCallResult = if result.processed_messages.is_empty() {
    { completion: result.completion, processed_messages: messages.to_owned(), }
  } else {
    result
  }
  Ok(final_result)
}

///|
pub impl Runtime for PortRuntime with fn execute_tool(self, ctx : EffectContext) -> @kernel.ToolOutcome {
  let provider = ctx.bound_tool
  let registration : ToolCancellationRegistration = {
    handler: None,
    active: true,
  }
  defer {
    registration.active = false
  }
  let register_cancel = fn(c : &@port.ToolCancellable) {
    if registration.active {
      registration.handler = Some(c)
    }
  }
  let call_ctx = (ctx.tool_context)(register_cancel)
  let run_id = ctx.run_id.to_string()
  let key = (run_id, ctx.effect_id.to_int())
  if self.cancels.contains(key) {
    raise @error.RuntimeError::InvocationFailed(
      "duplicate live tool effect: \{run_id}/\{ctx.effect_id.to_int()}",
    )
  }
  self.cancels[key] = registration
  let outcome = provider.execute(call_ctx) catch {
    error =>
      RuntimeFailure(
        error_category="ToolProviderError",
        message=error.to_string(),
      )
  }
  // Effect settled: drop the registration so a later cancel reports
  // NotPropagated instead of hitting a finished execution. Unwind teardown
  // of the effect task never reaches this line on purpose — there the
  // registration survives so a later cancel can still escalate the
  // orphaned child.
  self.cancels.remove(key)
  outcome
}

///|
/// Route cancellation to providers that registered a cancellable facet for
/// the effect. Unregistered ids — plain providers, or effects that already
/// settled — report `NotPropagated`, preserving the ports-only contract.
/// A facet failure propagates to core cleanup; its registration is retained
/// for shutdown retry rather than silently reported as a normal disposition.
pub impl Runtime for PortRuntime with fn cancel_effects(
  self : PortRuntime,
  run_id~,
  effect_ids : Array[@kernel.EffectId],
  reason : @kernel.CancelReason,
) -> Array[(@kernel.EffectId, @kernel.CancelDisposition)] {
  let dispositions : Array[(@kernel.EffectId, @kernel.CancelDisposition)] = []
  for id in effect_ids {
    let key = (run_id.to_string(), id.to_int())
    match self.cancels.get(key) {
      Some(registration) => {
        let disposition = match registration.handler {
          Some(handler) => handler.cancel(id, reason, run_id~)
          None => NotPropagated
        }
        if disposition is (Stopped | AlreadySettled) {
          self.cancels.remove(key)
        }
        dispositions.push((id, disposition))
      }
      None => dispositions.push((id, NotPropagated))
    }
  }
  dispositions
}

///|
pub impl Runtime for PortRuntime with fn compact(
  self,
  scope : @kernel.InvocationScope,
  messages : ArrayView[@kernel.Message],
  tools : Array[@kernel.ToolDef],
  call_options : Json,
  trigger : @kernel.CompactTrigger,
) -> Result[@kernel.CompactResult, @kernel.ModelFailure] {
  let options = match decode_chat_options(call_options) {
    Ok(options) => options
    Err(failure) => return Err(failure)
  }
  let result = self.model.compact(scope, messages, tools, options, trigger) catch {
    error => return Err(model_error_to_failure(error))
  }
  Ok(result)
}