Add AppKit governed effect surface

Author: nshkrdotcom

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge.ex

@@ -9,6 +9,7 @@ defmodule AppKit.Bridges.MezzanineBridge do
 
   alias AppKit.Bridges.MezzanineBridge.{
     AgentIntakeAdapter,
+    EffectAdapter,
     HeadlessAdapter,
     InstallationAdapter,
     OperatorAdapter,
@@ -28,6 +29,7 @@ defmodule AppKit.Bridges.MezzanineBridge do
   @behaviour AppKit.Core.Backends.HeadlessBackend
   @behaviour AppKit.Core.Backends.AgentIntakeBackend
   @behaviour AppKit.Core.Backends.RuntimeBackend
+  @behaviour AppKit.EffectSurface
 
   alias AppKit.Core.{
     AuthoringBundleImport,
@@ -468,4 +470,24 @@ defmodule AppKit.Bridges.MezzanineBridge do
   def await_agent_outcome(%RequestContext{} = context, run_ref, request, opts) do
     AgentIntakeAdapter.await_agent_outcome(context, run_ref, request, opts)
   end
+
+  @impl AppKit.EffectSurface
+  def propose_effect(%RequestContext{} = context, effect_params, opts) do
+    EffectAdapter.propose_effect(context, effect_params, opts)
+  end
+
+  @impl AppKit.EffectSurface
+  def get_effect(%RequestContext{} = context, effect_ref, opts) do
+    EffectAdapter.get_effect(context, effect_ref, opts)
+  end
+
+  @impl AppKit.EffectSurface
+  def list_effects(%RequestContext{} = context, run_ref, opts) do
+    EffectAdapter.list_effects(context, run_ref, opts)
+  end
+
+  @impl AppKit.EffectSurface
+  def get_effect_timeline(%RequestContext{} = context, effect_ref, opts) do
+    EffectAdapter.get_effect_timeline(context, effect_ref, opts)
+  end
 end

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/effect_adapter.ex

@@ -0,0 +1,110 @@
+defmodule AppKit.Bridges.MezzanineBridge.EffectAdapter do
+  @moduledoc false
+
+  @behaviour AppKit.EffectSurface
+
+  alias AppKit.Bridges.MezzanineBridge.{Errors, Services}
+
+  alias AppKit.Core.{
+    EffectTimelineDTO,
+    GovernedEffectDTO,
+    RequestContext
+  }
+
+  alias Mezzanine.Core.GovernedEffects.Coordinator
+  alias Mezzanine.Core.GovernedEffects.Coordinator.Run
+  alias Mezzanine.Core.GovernedEffects.Projection
+
+  @impl true
+  def propose_effect(%RequestContext{}, %GovernedEffectDTO{} = effect, opts)
+      when is_list(opts) do
+    effect
+    |> GovernedEffectDTO.dump()
+    |> propose_effect_from_attrs(opts)
+  end
+
+  def propose_effect(%RequestContext{}, effect_params, opts)
+      when is_map(effect_params) and is_list(opts) do
+    propose_effect_from_attrs(effect_params, opts)
+  end
+
+  @impl true
+  def get_effect(%RequestContext{}, effect_ref, opts)
+      when is_binary(effect_ref) and is_list(opts) do
+    case Services.effect_readback(opts).get_effect(effect_ref, opts) do
+      {:ok, %Run{} = run} -> dto_from_run(run)
+      {:error, reason} -> Errors.normalize(reason)
+    end
+  end
+
+  @impl true
+  def list_effects(%RequestContext{}, run_ref, opts) when is_binary(run_ref) and is_list(opts) do
+    case Services.effect_readback(opts).list_effects(run_ref, opts) do
+      {:ok, runs} -> collect_dtos(runs)
+      {:error, reason} -> Errors.normalize(reason)
+    end
+  end
+
+  @impl true
+  def get_effect_timeline(%RequestContext{}, effect_ref, opts)
+      when is_binary(effect_ref) and is_list(opts) do
+    with {:ok, %Run{} = run} <- Services.effect_readback(opts).get_effect(effect_ref, opts),
+         projection <- Projection.product_safe(run) do
+      EffectTimelineDTO.new(%{
+        effect_ref: effect_ref,
+        trace_summary_hash: Map.get(projection, "trace_summary_hash"),
+        entries: Map.get(projection, "timeline", [])
+      })
+    else
+      {:error, reason} -> Errors.normalize(reason)
+    end
+  end
+
+  defp propose_effect_from_attrs(attrs, opts) do
+    case Coordinator.propose(attrs, opts) do
+      {:ok, run} -> dto_from_run(run)
+      {:error, reason} -> Errors.normalize(reason)
+    end
+  end
+
+  defp collect_dtos(runs) do
+    Enum.reduce_while(runs, {:ok, []}, fn
+      %Run{} = run, {:ok, acc} ->
+        case dto_from_run(run) do
+          {:ok, dto} -> {:cont, {:ok, [dto | acc]}}
+          {:error, reason} -> {:halt, {:error, reason}}
+        end
+
+      _other, _acc ->
+        {:halt, Errors.normalize(:invalid_effect_projection)}
+    end)
+    |> case do
+      {:ok, dtos} -> {:ok, Enum.reverse(dtos)}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp dto_from_run(%Run{} = run) do
+    projection = Projection.product_safe(run)
+
+    GovernedEffectDTO.new(%{
+      effect_ref: run.effect.effect_ref,
+      effect_type: run.effect.effect_type,
+      command_ref: command_value(run.command, :command_ref),
+      tenant_ref: run.effect.tenant_ref,
+      actor_ref: run.effect.actor_ref,
+      installation_ref: run.effect.installation_ref,
+      status: Map.get(projection, "status"),
+      trace_ref: run.effect.trace_ref,
+      authority_ref: run.effect.authority_ref,
+      receipt_ref: run.effect.receipt_ref,
+      dispatch_ref: run.effect.dispatch_ref,
+      expected_version: run.effect.expected_version,
+      metadata: %{
+        "trace_summary_hash" => Map.get(projection, "trace_summary_hash")
+      }
+    })
+  end
+
+  defp command_value(command, key), do: Map.get(command, key, Map.get(command, to_string(key)))
+end

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/runtime_adapter.ex

@@ -46,6 +46,7 @@ defmodule AppKit.Bridges.MezzanineBridge.RuntimeAdapter do
              accepted?: true,
              command_ref: "command://#{agent_request.idempotency_key}",
              correlation_id: agent_request.correlation_id,
+             governed_effect_refs: RuntimeMapping.governed_effect_refs(projection, agent_request),
              polling_hint: %{checking?: false, poll_interval_ms: 1_000, staleness_ms: 0}
            }) do
       {:ok, future}

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/runtime_mapping.ex

@@ -61,6 +61,9 @@ defmodule AppKit.Bridges.MezzanineBridge.RuntimeMapping do
        max_turns: param(params, :max_turns, 1),
        timeout_policy: timeout_policy(params),
        profile_bundle: Map.from_struct(profile_bundle),
+       effect_governance_mode: request.effect_governance_mode,
+       diagnostic_lane: request.diagnostic_lane,
+       governed_effect_refs: request.governed_effect_refs,
        fixture_script: param(params, :fixture_script, "success_first_try"),
        continue_as_new_turn_threshold: param(params, :continue_as_new_turn_threshold, 50),
        source_ref: "actor://#{context.actor_ref.id}"
@@ -120,6 +123,13 @@ defmodule AppKit.Bridges.MezzanineBridge.RuntimeMapping do
       :session_turn
   end
 
+  def governed_effect_refs(projection, %AgentRunRequest{} = request) do
+    case Common.fetch_value(projection, :governed_effect_refs) do
+      refs when is_map(refs) and map_size(refs) > 0 -> refs
+      _missing -> request.governed_effect_refs || %{}
+    end
+  end
+
   def runtime_available?(runtime) when is_atom(runtime),
     do: Code.ensure_loaded?(runtime) and function_exported?(runtime, :run, 1)
 

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/services.ex

@@ -25,6 +25,9 @@ defmodule AppKit.Bridges.MezzanineBridge.Services do
   def runtime_gateway(opts),
     do: Keyword.get(opts, :runtime_gateway_service, Mezzanine.AppKitBridge.RuntimeGatewayService)
 
+  def effect_readback(opts),
+    do: Keyword.get(opts, :effect_readback_service, Mezzanine.AppKitBridge.EffectReadbackService)
+
   def program_context(opts),
     do: Keyword.get(opts, :program_context_service, Mezzanine.AppKitBridge.ProgramContextService)
 

bridges/mezzanine_bridge/lib/mezzanine/app_kit_bridge/effect_readback_service.ex

@@ -0,0 +1,30 @@
+defmodule Mezzanine.AppKitBridge.EffectReadbackService do
+  @moduledoc false
+
+  alias Mezzanine.Core.GovernedEffects.Coordinator.Run
+
+  @spec get_effect(String.t(), keyword()) :: {:ok, Run.t()} | {:error, term()}
+  def get_effect(effect_ref, opts) when is_binary(effect_ref) and is_list(opts) do
+    case run_for(effect_ref, opts) do
+      %Run{} = run -> {:ok, run}
+      nil -> {:error, :effect_projection_not_configured}
+    end
+  end
+
+  @spec list_effects(String.t(), keyword()) :: {:ok, [Run.t()]} | {:error, term()}
+  def list_effects(_run_ref, opts) when is_list(opts) do
+    runs =
+      opts
+      |> Keyword.get(:effect_runs, %{})
+      |> Map.values()
+
+    {:ok, Enum.filter(runs, &match?(%Run{}, &1))}
+  end
+
+  defp run_for(effect_ref, opts) do
+    case Keyword.get(opts, :effect_runs, %{}) do
+      runs when is_map(runs) -> Map.get(runs, effect_ref)
+      _other -> nil
+    end
+  end
+end

bridges/mezzanine_bridge/mix.exs

@@ -70,6 +70,7 @@ defmodule AppKitMezzanineBridge.MixProject do
       DependencySources.dep(:mezzanine_projection_engine, @repo_root),
       DependencySources.dep(:mezzanine_source_engine, @repo_root),
       DependencySources.dep(:mezzanine_decision_engine, @repo_root),
+      DependencySources.dep(:mezzanine_governed_effects, @repo_root),
       DependencySources.dep(:mezzanine_evidence_engine, @repo_root),
       DependencySources.dep(:mezzanine_archival_engine, @repo_root),
       DependencySources.dep(:mezzanine_config_registry, @repo_root),

bridges/mezzanine_bridge/test/app_kit/bridges/mezzanine_bridge_contract_characterization_test.exs

@@ -176,6 +176,17 @@ defmodule AppKit.Bridges.MezzanineBridgeContractCharacterizationTest do
         cancel_agent_run: 3,
         await_agent_outcome: 4
       ]
+    },
+    %{
+      behaviour: AppKit.EffectSurface,
+      target_module: "AppKit.Bridges.MezzanineBridge.EffectAdapter",
+      service_options: [:effect_readback],
+      callbacks: [
+        propose_effect: 3,
+        get_effect: 3,
+        list_effects: 3,
+        get_effect_timeline: 3
+      ]
     }
   ]
 
@@ -740,7 +751,8 @@ defmodule AppKit.Bridges.MezzanineBridgeContractCharacterizationTest do
              "AppKit.Bridges.MezzanineBridge.OperatorAdapter",
              "AppKit.Bridges.MezzanineBridge.RuntimeAdapter",
              "AppKit.Bridges.MezzanineBridge.HeadlessAdapter",
-             "AppKit.Bridges.MezzanineBridge.AgentIntakeAdapter"
+             "AppKit.Bridges.MezzanineBridge.AgentIntakeAdapter",
+             "AppKit.Bridges.MezzanineBridge.EffectAdapter"
            ]
 
     for adapter <- @adapter_contract, {callback, arity} <- adapter.callbacks do

bridges/mezzanine_bridge/test/app_kit/bridges/mezzanine_bridge_effect_adapter_test.exs

@@ -0,0 +1,66 @@
+defmodule AppKit.Bridges.MezzanineBridgeEffectAdapterTest do
+  use ExUnit.Case, async: true
+
+  alias AppKit.Bridges.MezzanineBridge.EffectAdapter
+  alias AppKit.Core.{EffectTimelineDTO, GovernedEffectDTO, RequestContext}
+
+  test "effect adapter delegates proposal to Mezzanine governed-effect coordinator" do
+    assert {:ok, %GovernedEffectDTO{} = dto} =
+             EffectAdapter.propose_effect(context!(), effect_attrs(), [])
+
+    assert dto.effect_ref == "effect://tenant-1/effects/1"
+    assert dto.status == "proposed"
+    assert dto.trace_ref == "trace://tenant-1/effects/1"
+  end
+
+  test "effect adapter readback delegates projection through Mezzanine run data" do
+    assert {:ok, run} =
+             Mezzanine.Core.GovernedEffects.Coordinator.propose(effect_attrs())
+
+    opts = [effect_runs: %{run.effect.effect_ref => run}]
+
+    assert {:ok, %GovernedEffectDTO{effect_ref: "effect://tenant-1/effects/1"}} =
+             EffectAdapter.get_effect(context!(), "effect://tenant-1/effects/1", opts)
+
+    assert {:ok, [%GovernedEffectDTO{effect_ref: "effect://tenant-1/effects/1"}]} =
+             EffectAdapter.list_effects(context!(), "run://tenant-1/runs/1", opts)
+
+    assert {:ok, %EffectTimelineDTO{entries: [%{"event_kind" => "effect_transition"}]}} =
+             EffectAdapter.get_effect_timeline(context!(), "effect://tenant-1/effects/1", opts)
+  end
+
+  defp context! do
+    {:ok, context} =
+      RequestContext.new(%{
+        trace_id: "0123456789abcdef0123456789abcdef",
+        actor_ref: %{
+          id: "actor://tenant-1/operator",
+          kind: "operator",
+          roles: ["developer"]
+        },
+        tenant_ref: %{id: "tenant://tenant-1", slug: "tenant-1"},
+        installation_ref: %{
+          id: "installation://tenant-1/synapse",
+          pack_slug: "synapse"
+        },
+        request_id: "request-1",
+        idempotency_key: "idempotency://tenant-1/request-1"
+      })
+
+    context
+  end
+
+  defp effect_attrs do
+    %{
+      effect_ref: "effect://tenant-1/effects/1",
+      effect_type: "diagnostic.echo",
+      command_ref: "command://tenant-1/commands/1",
+      tenant_ref: "tenant://tenant-1",
+      actor_ref: "actor://tenant-1/operator",
+      installation_ref: "installation://tenant-1/synapse",
+      trace_ref: "trace://tenant-1/effects/1",
+      expected_version: 1,
+      payload: %{"message_ref" => "payload://tenant-1/messages/1"}
+    }
+  end
+end

build_support/dependency_sources.config.exs

@@ -11,6 +11,17 @@
       default_order: [:path, :github, :hex],
       publish_order: [:hex]
     },
+    mezzanine_governed_effects: %{
+      path: "../mezzanine/core/governed_effects",
+      hex: "~> 0.1.0",
+      github: %{
+        repo: "nshkrdotcom/mezzanine",
+        subdir: "core/governed_effects",
+        branch: "main"
+      },
+      default_order: [:path, :github, :hex],
+      publish_order: [:hex]
+    },
     mezzanine_projection_engine: %{
       path: "../mezzanine/core/projection_engine",
       hex: "~> 0.1.0",