okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffix(extensions): activate bundles atomically with dependencies52bd5192026-10-07

//// A broken extension must cost only itself. Albedo loads third-party
//// extensions into the same session as its own, and a plugin that fails or
//// raises used to take the whole session with it: no tools, no turn, no
//// reply, and for a tool error a dead subagent. E2E cannot reach this,
//// because the daemon only loads albedo's own registry, so the rule is
//// checked here against deliberately broken extensions.

import albedo/daemon/conversation
import albedo/daemon/family
import albedo/daemon/store
import albedo/harness/extension
import albedo/harness/extension/composition
import albedo/harness/extensions
import albedo/harness/runtime
import albedo/harness/tool
import albedo/openai_api/types
import gleam/dynamic/decode
import gleam/erlang/process
import gleam/json
import gleam/list
import gleam/option.{None, Some}
import gleam/string
import gleeunit/should

/// A session over `installed`, every one of them enabled.
fn host(
  installed: List(extension.Extension),
) -> #(runtime.Runtime, runtime.Session) {
  let assert Ok(host) =
    runtime.start_with_config(
      ":memory:",
      extensions.Config(
        installed,
        list.map(installed, fn(extension) { extension.name }),
      ),
    )
  let ledger = runtime.ledger(host)
  let assert Ok(_) = conversation.initialise(ledger)
  let assert Ok(_) = family.initialise(ledger)
  let assert Ok(_) =
    conversation.create(
      ledger,
      conversation.Info(
        "broken-extension-test",
        "new session",
        "/tmp",
        "provider",
        "model",
        types.Responses,
        conversation.Idle,
        None,
        None,
      ),
    )
  let assert Ok(session) =
    runtime.open_session(host, "broken-extension-test", "/tmp")
  #(host, session)
}

/// An extension whose only tool answers, named after it.
fn healthy(name: String) -> extension.Extension {
  extension.Extension(
    name,
    "answers",
    [],
    [extension.ToolPlugin("", [answering(name)], [], [])],
    extension.no_initialise,
  )
}

fn answering(name: String) -> extension.Tool {
  tool.text(
    name,
    "answers",
    False,
    [],
    [],
    decode.success(Nil),
    "usage",
    fn(_, _) { Ok("answered") },
  )
}

fn tool_only(name: String, value: extension.Tool) -> extension.Extension {
  extension.Extension(
    name,
    "one tool",
    [],
    [extension.ToolPlugin("", [value], [], [])],
    extension.no_initialise,
  )
}

fn plugin_only(name: String, plugin: extension.Plugin) -> extension.Extension {
  extension.Extension(name, "one plugin", [], [plugin], extension.no_initialise)
}

fn tool_names(session: runtime.Session) -> List(String) {
  runtime.tools(session) |> list.map(fn(tool) { tool.name })
}

fn warned_about(session: runtime.Session, name: String) -> Bool {
  runtime.warnings(session)
  |> list.any(fn(warning) { string.contains(warning, name) })
}

fn answer(
  host: runtime.Runtime,
  session: runtime.Session,
  name: String,
) -> String {
  let assert Ok(types.ToolOutput(_, body, _)) =
    runtime.invoke(
      host,
      session,
      types.ToolCall("call_1", name, "{}"),
      types.any_images,
    )
  body
}

pub fn a_failed_or_crashed_context_load_keeps_the_session_test() -> Nil {
  let #(host, session) =
    host([
      plugin_only("sulky", extension.ContextPlugin(fn(_) { Error("no root") })),
      plugin_only("exploding", extension.ContextPlugin(fn(_) { panic as "bug" })),
      healthy("working"),
    ])
  tool_names(session) |> should.equal(["working"])
  warned_about(session, "sulky") |> should.be_true
  warned_about(session, "exploding") |> should.be_true
  runtime.stop(host)
}

pub fn a_failed_or_crashed_prepare_keeps_the_session_test() -> Nil {
  let #(host, session) =
    host([
      plugin_only(
        "sulky",
        extension.ManagedPlugin(fn(_, _, _) { Error("no server") }),
      ),
      plugin_only(
        "exploding",
        extension.ManagedPlugin(fn(_, _, _) { panic as "bug" }),
      ),
      healthy("working"),
    ])
  tool_names(session) |> should.equal(["working"])
  warned_about(session, "sulky") |> should.be_true
  warned_about(session, "exploding") |> should.be_true
  runtime.stop(host)
}

/// A static collision is refused when the extension is selected; one a
/// managed plugin only discovers while preparing, as an MCP server's tool
/// list does, reaches the session and must lose there.
pub fn a_colliding_extension_loses_only_itself_test() -> Nil {
  let #(host, session) =
    host([
      healthy("echo"),
      plugin_only(
        "squatter",
        extension.ManagedPlugin(fn(_, _, _) {
          Ok(extension.Managed(..extension.empty(), tools: [answering("echo")]))
        }),
      ),
    ])
  tool_names(session) |> should.equal(["echo"])
  warned_about(session, "squatter") |> should.be_true
  answer(host, session, "echo") |> should.equal("answered")
  runtime.stop(host)
}

pub fn a_crashing_tool_answers_instead_of_ending_the_turn_test() -> Nil {
  let #(host, session) =
    host([
      tool_only(
        "exploding",
        extension.Tool(
          types.Tool("boom", "crashes", tool.schema([], []), False),
          fn(_, _) { panic as "tool bug" },
          fn(_) { None },
        ),
      ),
    ])
  let reason = error_of(answer(host, session, "boom"))
  string.contains(reason, "boom crashed") |> should.be_true
  string.contains(reason, "inspect") |> should.be_true
  runtime.stop(host)
}

pub fn a_fatal_tool_still_ends_the_turn_test() -> Nil {
  let #(host, session) =
    host([
      tool_only(
        "strict",
        extension.Tool(
          types.Tool("record", "cannot record", tool.schema([], []), False),
          fn(_, _) { Error(extension.Fatal("transcript storage failed")) },
          fn(_) { None },
        ),
      ),
    ])
  runtime.invoke(
    host,
    session,
    types.ToolCall("call_1", "record", "{}"),
    types.any_images,
  )
  |> should.equal(Error("transcript storage failed"))
  runtime.stop(host)
}

pub fn a_crashing_observer_leaves_the_session_working_test() -> Nil {
  let #(host, session) =
    host([
      plugin_only(
        "exploding",
        extension.ManagedPlugin(fn(_, _, _) {
          Ok(
            extension.Managed(..extension.empty(), observe: fn(_, _) {
              panic as "observer bug"
            }),
          )
        }),
      ),
      healthy("working"),
    ])
  runtime.observe(
    session,
    extension.Session(
      "broken-extension-test",
      fn(_, _) { Error("no upstream") },
      fn(_) { Nil },
      fn() { False },
    ),
    extension.Stirred,
  )
  answer(host, session, "working") |> should.equal("answered")
  runtime.stop(host)
}

fn error_of(body: String) -> String {
  let assert Ok(reason) =
    json.parse(body, decode.field("error", decode.string, decode.success))
  reason
}

/// An extension whose initialiser `fails`, for the quarantine cases.
fn uninstallable(name: String, fails: fn() -> Nil) -> extension.Extension {
  extension.Extension(
    name,
    "will not install",
    [],
    [extension.ToolPlugin("", [answering(name)], [], [])],
    fn(_) {
      fails()
      Error("no tables")
    },
  )
}

fn summary(
  host: runtime.Runtime,
  name: String,
) -> Result(extension.Summary, Nil) {
  let assert Ok(summaries) =
    runtime.extension_summaries(host, "broken-extension-test")
  list.find(summaries, fn(summary) { summary.name == name })
}

pub fn an_extension_that_cannot_install_is_quarantined_test() -> Nil {
  let #(host, session) =
    host([
      uninstallable("sulky", fn() { Nil }),
      uninstallable("exploding", fn() { panic as "initialiser bug" }),
      healthy("working"),
    ])
  tool_names(session) |> should.equal(["working"])
  let assert Ok(extension.Summary(quarantined: Some(reason), enabled: False, ..)) =
    summary(host, "sulky")
  string.contains(reason, "no tables") |> should.be_true
  let assert Ok(extension.Summary(quarantined: Some(crash), ..)) =
    summary(host, "exploding")
  string.contains(crash, "initialiser bug") |> should.be_true
  runtime.stop(host)
}

pub fn a_quarantined_requirement_takes_its_dependents_with_it_test() -> Nil {
  let #(host, session) =
    host([
      uninstallable("engine", fn() { Nil }),
      extension.Extension(
        "rider",
        "needs the engine",
        ["engine"],
        [extension.ToolPlugin("", [answering("rider")], [], [])],
        extension.no_initialise,
      ),
      healthy("working"),
    ])
  tool_names(session) |> should.equal(["working"])
  let assert Ok(extension.Summary(quarantined: Some(reason), ..)) =
    summary(host, "rider")
  string.contains(reason, "it requires engine") |> should.be_true
  runtime.stop(host)
}

pub fn a_quarantined_extension_cannot_be_enabled_test() -> Nil {
  let #(host, _) =
    host([uninstallable("sulky", fn() { Nil }), healthy("working")])
  let assert Error(refusal) =
    runtime.reload_extension(
      host,
      "broken-extension-test",
      "/tmp",
      "sulky",
      True,
    )
  string.contains(refusal, "quarantined") |> should.be_true
  runtime.stop(host)
}

/// Mixed plugins and transitive failures require synthetic registry entries;
/// the daemon only installs its compiled registry.
pub fn a_failed_bundle_removes_all_capabilities_and_dependents_test() -> Nil {
  let assert Ok(ledger) = store.start(":memory:", "")
  let closed = process.new_subject()
  let managed = fn(name) {
    extension.ManagedPlugin(fn(_, _, _) {
      Ok(
        extension.Managed(
          ..extension.empty(),
          tools: [answering(name)],
          close: fn() { process.send(closed, name) },
        ),
      )
    })
  }
  let extensions = [
    extension.Extension(
      "dependent",
      "dependent",
      ["broken"],
      [managed("dependent")],
      extension.no_initialise,
    ),
    extension.Extension(
      "transitive",
      "transitive",
      ["dependent"],
      [managed("transitive")],
      extension.no_initialise,
    ),
    extension.Extension(
      "broken",
      "mixed",
      [],
      [
        extension.ToolPlugin("must disappear", [answering("static")], [], []),
        managed("prepared"),
        extension.ManagedPlugin(fn(_, _, _) { Error("prepare refused") }),
      ],
      extension.no_initialise,
    ),
    healthy("working"),
  ]
  let composed = composition.compose(extensions, ledger, "s", "/tmp")
  composition.tools(composed)
  |> list.map(fn(tool) { tool.definition.name })
  |> should.equal(["working"])
  composition.extensions(composed)
  |> list.map(fn(item) { item.name })
  |> should.equal(["working"])
  composition.requested(composed) |> list.length |> should.equal(4)
  composition.inactive(composed) |> list.length |> should.equal(3)
  composition.instructions(composed) |> should.equal("")
  process.receive(closed, 100) |> should.equal(Ok("prepared"))
  process.receive(closed, 100) |> should.equal(Ok("transitive"))
  process.receive(closed, 100) |> should.equal(Ok("dependent"))
  composition.close(composed)
  process.receive(closed, 0) |> should.equal(Error(Nil))
  store.close(ledger)
}

pub fn context_failure_never_prepares_the_rest_of_the_bundle_test() -> Nil {
  let assert Ok(ledger) = store.start(":memory:", "")
  let prepared = process.new_subject()
  let composed =
    composition.compose(
      [
        extension.Extension(
          "broken",
          "context",
          [],
          [
            extension.ContextPlugin(fn(_) { Error("context refused") }),
            extension.ManagedPlugin(fn(_, _, _) {
              process.send(prepared, Nil)
              Ok(extension.empty())
            }),
          ],
          extension.no_initialise,
        ),
      ],
      ledger,
      "s",
      "/tmp",
    )
  composition.extensions(composed) |> should.equal([])
  process.receive(prepared, 0) |> should.equal(Error(Nil))
  composition.close(composed)
  store.close(ledger)
}

pub fn registry_order_resolves_whole_bundle_collisions_test() -> Nil {
  let assert Ok(ledger) = store.start(":memory:", "")
  let closed = process.new_subject()
  let composed =
    composition.compose(
      [
        plugin_only(
          "first",
          extension.ManagedPlugin(fn(_, _, _) {
            Ok(
              extension.Managed(..extension.empty(), tools: [
                answering("claimed"),
              ]),
            )
          }),
        ),
        extension.Extension(
          "loser",
          "mixed",
          [],
          [
            extension.ToolPlugin(
              "",
              [answering("claimed"), answering("extra")],
              [],
              [],
            ),
            extension.ManagedPlugin(fn(_, _, _) {
              Ok(
                extension.Managed(..extension.empty(), close: fn() {
                  process.send(closed, Nil)
                }),
              )
            }),
          ],
          extension.no_initialise,
        ),
      ],
      ledger,
      "s",
      "/tmp",
    )
  composition.tools(composed)
  |> list.map(fn(tool) { tool.definition.name })
  |> should.equal(["claimed"])
  composition.extensions(composed)
  |> list.map(fn(item) { item.name })
  |> should.equal(["first"])
  process.receive(closed, 100) |> should.equal(Ok(Nil))
  composition.close(composed)
  process.receive(closed, 0) |> should.equal(Error(Nil))
  store.close(ledger)
}