test/openagents/turn_provenance_test.exs

main at 58e6347eeb72 · 14 KB

defmodule OpenAgents.TurnProvenanceTest do
  use OpenAgents.DataCase
  alias OpenAgents.{Context.Composer, Conversations, Turns}
  alias OpenAgents.Conversations.{ProviderStep, TurnReceipt}
  alias OpenAgents.Provenance.Canonical
  alias OpenAgents.Providers.Request

  test "captures immutable identity before provider work starts" do
    %{request: request, context: context, receipt: receipt, provider_step: step, turn: turn} =
      begin_turn("capture-browser", "Explain this.")

    assert turn.status == "streaming"
    assert receipt.status == "captured"
    assert receipt.schema_version == 1
    assert receipt.model_id == request.model_id
    assert receipt.persona_id == context.persona_id
    assert receipt.persona_digest == context.persona_digest
    assert receipt.role_id == context.role_id
    assert receipt.role_digest == context.role_digest
    assert receipt.role_selection == context.role_selection
    assert receipt.role_selection["schema"] == "sarah.role_selection.v1"
    assert receipt.role_selection["reason"] == "public_default"
    assert receipt.role_selection["surface"] == "text"
    assert receipt.instruction_digest == context.instruction_digest
    assert receipt.input_digest == Canonical.digest!(request.input)
    assert receipt.input_message_count == length(request.input)
    assert receipt.input_bytes == byte_size(Canonical.encode!(request.input))
    assert receipt.used_source_refs == []
    assert receipt.used_tool_step_refs == []

    assert receipt.used_memory_evidence == %{
             "schema" => "sarah.memory_evidence_usage.v1",
             "items" => []
           }

    assert receipt.tool_catalog_digest == nil
    assert receipt.blueprint_revision == nil
    assert receipt.program_artifact_id == "sarah.program.memory_intent.shadow.v1"
    assert byte_size(receipt.program_artifact_digest) == 64
    assert receipt.program_artifact_receipt["schema"] == "sarah.program_capture.v1"
    refute receipt.program_artifact_receipt["degraded"]
    assert receipt.memory_snapshot_ref =~ ~r/^message:[0-9a-f-]{36}$/
    assert step.sequence == 1
    assert step.provider_id == "test.provider"
    assert step.model_id == request.model_id
    assert step.status == "started"
  end

  test "reconstructs ordered provider responses, usage, and used references" do
    %{receipt: receipt, turn: turn} = begin_turn("chain-browser", "Use two provider steps.")

    assert {:ok, completed_first_step} =
             Conversations.record_provider_step_completion(
               receipt,
               "response-1",
               %{"input_tokens" => 10, "output_tokens" => 2}
             )

    assert completed_first_step.sequence == 1

    assert {:ok, second_step} =
             Conversations.start_provider_step(receipt, "test.provider", "model-v2")

    assert second_step.sequence == 2

    assert {:ok, updated_receipt} =
             Conversations.record_used_refs(receipt,
               source_refs: ["message:2", "message:1", "message:1"],
               tool_step_refs: ["tool-step:1"]
             )

    assert updated_receipt.used_source_refs == ["message:1", "message:2"]
    assert updated_receipt.used_tool_step_refs == ["tool-step:1"]

    assert {:ok, completed_turn} =
             Conversations.complete_turn(
               turn,
               "response-2",
               %{"input_tokens" => 4, "output_tokens" => 8}
             )

    assert completed_turn.status == "completed"
    assert {:ok, terminal_receipt} = Conversations.get_turn_receipt(completed_turn)
    assert terminal_receipt.status == "completed"
    assert terminal_receipt.provider_completed_at
    assert terminal_receipt.usage == %{"input_tokens" => 4, "output_tokens" => 8}

    assert [first_step, terminal_second_step] =
             Conversations.list_provider_steps(terminal_receipt)

    assert first_step.provider_response_id == "response-1"
    assert first_step.usage == %{"input_tokens" => 10, "output_tokens" => 2}
    assert terminal_second_step.sequence == 2
    assert terminal_second_step.provider_response_id == "response-2"
    assert terminal_second_step.status == "completed"

    assert {:error, :turn_receipt_is_terminal} =
             Conversations.record_used_refs(terminal_receipt,
               source_refs: ["message:after-terminal"]
             )
  end

  test "host failure after a completed provider response preserves that response step" do
    %{receipt: receipt, turn: turn} = begin_turn("host-failure-browser", "Finish provider first.")

    assert {:ok, %ProviderStep{status: "completed"}} =
             Conversations.record_provider_step_completion(receipt, "provider-finished")

    assert {:ok, failed_turn} = Conversations.fail_turn(turn, :host_persistence_failed)
    assert {:ok, failed_receipt} = Conversations.get_turn_receipt(failed_turn)
    assert failed_receipt.status == "failed"

    assert [%ProviderStep{status: "completed", provider_response_id: "provider-finished"}] =
             Conversations.list_provider_steps(failed_receipt)
  end

  test "failure and cancellation retain captured provenance" do
    failed = begin_turn("failed-browser", "Fail with provenance.")
    assert {:ok, failed_turn} = Conversations.fail_turn(failed.turn, {:provider_error, "private"})
    assert {:ok, failed_receipt} = Conversations.get_turn_receipt(failed_turn)
    assert failed_receipt.status == "failed"
    assert failed_receipt.persona_digest == failed.context.persona_digest

    assert [%ProviderStep{status: "failed", error_code: "provider_error"}] =
             Conversations.list_provider_steps(failed_receipt)

    cancelled = begin_turn("cancelled-browser", "Cancel with provenance.")
    assert {:ok, cancelled_turn} = Conversations.cancel_turn(cancelled.turn)
    assert {:ok, cancelled_receipt} = Conversations.get_turn_receipt(cancelled_turn)
    assert cancelled_receipt.status == "cancelled"

    assert [%ProviderStep{status: "cancelled", error_code: "cancelled"}] =
             Conversations.list_provider_steps(cancelled_receipt)
  end

  test "startup recovery marks partial receipts and steps interrupted" do
    %{receipt: receipt} = begin_turn("interrupted-browser", "Do not lose provenance.")

    assert :ok = Conversations.recover_interrupted_turns()

    recovered_receipt = Repo.get!(TurnReceipt, receipt.id)
    assert recovered_receipt.status == "interrupted"
    assert recovered_receipt.provider_completed_at

    assert [%ProviderStep{status: "interrupted", error_code: "runtime_restarted"}] =
             Conversations.list_provider_steps(recovered_receipt)
  end

  test "TurnRecovery finalizes durable evidence after the live server dies mid-stream" do
    Application.put_env(:openagents, :test_provider_observer, self())
    on_exit(fn -> Application.delete_env(:openagents, :test_provider_observer) end)

    assert {:ok, conversation} = Conversations.ensure_conversation("turn-worker-recovery")
    assert {:ok, records} = Conversations.create_turn(conversation, "[observe-request]")
    assert {:ok, turn_server} = Turns.start(records.turn.id)
    assert_receive {:provider_request, provider_task, _request}

    server_ref = Process.monitor(turn_server)
    Process.exit(turn_server, :kill)
    assert_receive {:DOWN, ^server_ref, :process, ^turn_server, :killed}
    provider_ref = Process.monitor(provider_task)
    send(provider_task, :continue_provider)
    assert_receive {:DOWN, ^provider_ref, :process, ^provider_task, _reason}, 1_000

    recovery = start_supervised!({OpenAgents.TurnRecovery, []})
    _state = :sys.get_state(recovery)

    recovered_turn = Conversations.get_turn!(records.turn.id)
    recovered_message = Repo.get!(OpenAgents.Conversations.Message, records.assistant_message.id)
    {:ok, recovered_receipt} = Conversations.get_turn_receipt(recovered_turn)

    assert recovered_turn.status == "failed"
    assert recovered_turn.error_code == "runtime_restarted"
    assert recovered_message.status == "failed"
    assert recovered_receipt.status == "interrupted"

    assert [%ProviderStep{status: "interrupted", error_code: "runtime_restarted"}] =
             Conversations.list_provider_steps(recovered_receipt)
  end

  test "legacy turns remain explicit instead of receiving fabricated provenance" do
    assert {:ok, conversation} = Conversations.ensure_conversation("legacy-browser")
    assert {:ok, records} = Conversations.create_turn(conversation, "Legacy turn.")

    assert {:error, :legacy_turn_without_receipt} =
             Conversations.get_turn_receipt(records.turn)

    assert {:ok, completed_turn} = Conversations.complete_turn(records.turn, "legacy-response")
    assert completed_turn.status == "completed"

    assert {:error, :legacy_turn_without_receipt} =
             Conversations.get_turn_receipt(completed_turn)
  end

  test "the database rejects mid-turn identity mutation" do
    %{receipt: receipt} = begin_turn("immutable-browser", "Keep identity fixed.")

    assert_raise Postgrex.Error, fn ->
      Repo.update_all(
        from(stored_receipt in TurnReceipt, where: stored_receipt.id == ^receipt.id),
        set: [persona_id: "sarah.persona.changed"]
      )
    end
  end

  test "the database rejects role-selection provenance mutation" do
    %{receipt: receipt} = begin_turn("immutable-role-selection", "Keep the selected role fixed.")

    assert_raise Postgrex.Error, fn ->
      Repo.update_all(
        from(stored_receipt in TurnReceipt, where: stored_receipt.id == ^receipt.id),
        set: [role_selection: Map.put(receipt.role_selection, "reason", "silent_sales")]
      )
    end
  end

  test "records an explicit degraded program baseline and protects its receipt" do
    catalog = OpenAgents.ProgramArtifacts.current!()
    degraded = OpenAgents.ProgramArtifacts.capture(catalog, "sarah.unknown.signature.v1")

    %{receipt: receipt} =
      begin_turn(
        "program-degraded-browser",
        "Use the deterministic baseline.",
        "model-v1",
        program_snapshot: degraded
      )

    assert receipt.program_artifact_id == nil
    assert receipt.program_artifact_digest == nil
    assert receipt.program_artifact_receipt["degraded"]
    assert receipt.program_artifact_receipt["reason"] =~ "deterministic_baseline"

    assert_raise Postgrex.Error, fn ->
      Repo.update_all(
        from(stored_receipt in TurnReceipt, where: stored_receipt.id == ^receipt.id),
        set: [
          program_artifact_receipt: Map.put(receipt.program_artifact_receipt, "degraded", false)
        ]
      )
    end
  end

  test "the database keeps the captured profile-memory snapshot immutable" do
    snapshot_ref = "profile-memory-snapshot:v1:#{Ecto.UUID.generate()}"

    %{receipt: receipt} =
      begin_turn("immutable-profile-snapshot", "Keep memory fixed.", "model-v1",
        profile_memory_snapshot_ref: snapshot_ref
      )

    assert receipt.profile_memory_snapshot_ref == snapshot_ref

    assert_raise Postgrex.Error, fn ->
      Repo.update_all(
        from(stored_receipt in TurnReceipt, where: stored_receipt.id == ^receipt.id),
        set: [profile_memory_snapshot_ref: "profile-memory-snapshot:v1:#{Ecto.UUID.generate()}"]
      )
    end
  end

  test "the database rejects terminal memory evidence mutation" do
    %{receipt: receipt, turn: turn} = begin_turn("terminal-evidence-browser", "Freeze evidence.")

    assert {:ok, _completed_turn} =
             Conversations.complete_turn(turn, "terminal-evidence-response")

    assert_raise Postgrex.Error, fn ->
      Repo.update_all(
        from(stored_receipt in TurnReceipt, where: stored_receipt.id == ^receipt.id),
        set: [
          used_memory_evidence: %{
            "schema" => "sarah.memory_evidence_usage.v1",
            "items" => [
              %{
                "source_ref" => "message:#{Ecto.UUID.generate()}",
                "classification" => "applicable"
              }
            ]
          }
        ]
      )
    end
  end

  test "a model configuration change affects only later turns" do
    original_model = Application.fetch_env!(:openagents, :openai_model)
    Application.put_env(:openagents, :test_provider_observer, self())

    on_exit(fn ->
      Application.put_env(:openagents, :openai_model, original_model)
      Application.delete_env(:openagents, :test_provider_observer)
    end)

    first = start_observed_turn("config-browser-one")
    assert_receive {:provider_request, first_task, first_request}
    assert first_request.model_id == original_model

    Application.put_env(:openagents, :openai_model, "model-after-cutover")
    first_pid = first.pid
    first_monitor = Process.monitor(first_pid)
    send(first_task, :continue_provider)
    assert_receive {:DOWN, ^first_monitor, :process, ^first_pid, :normal}

    assert {:ok, first_receipt} =
             first.turn_id |> Conversations.get_turn!() |> Conversations.get_turn_receipt()

    assert first_receipt.model_id == original_model
    assert first_receipt.tool_catalog_digest == OpenAgents.Tools.Registry.current!().digest

    second = start_observed_turn("config-browser-two")
    assert_receive {:provider_request, second_task, second_request}
    assert second_request.model_id == "model-after-cutover"
    second_pid = second.pid
    second_monitor = Process.monitor(second_pid)
    send(second_task, :continue_provider)
    assert_receive {:DOWN, ^second_monitor, :process, ^second_pid, :normal}
  end

  defp begin_turn(browser_key, content, model_id \\ "model-v1", options \\ []) do
    assert {:ok, conversation} = Conversations.ensure_conversation(browser_key)
    assert {:ok, records} = Conversations.create_turn(conversation, content)
    context = Composer.compose!()
    messages = Conversations.provider_messages(conversation.id)
    request = %Request{model_id: model_id, instructions: context.instructions, input: messages}

    assert {:ok, inference} =
             Conversations.begin_inference(
               records.turn,
               context,
               request,
               "test.provider",
               options
             )

    Map.merge(inference, %{context: context, request: request})
  end

  defp start_observed_turn(browser_key) do
    assert {:ok, conversation} = Conversations.ensure_conversation(browser_key)
    assert {:ok, records} = Conversations.create_turn(conversation, "[observe-request]")
    assert {:ok, pid} = Turns.start(records.turn.id)
    %{pid: pid, turn_id: records.turn.id}
  end
end