Host import
This notebook decodes the shipped artifacts, compares declared build identities with host observations, intersects bounds, calls reconcile/3, and reads each part of the resulting evidence.
The first cell installs the published package at the release candidate requirement.
Mix.install([{:agent_blueprint_protocol, "~> 1.0.0-rc.2"}])Locate and decode both shipped artifacts and the plain JSON views used to build host observations.
dependency_path =
try do
Mix.Project.deps_paths() |> Map.get(:agent_blueprint_protocol)
rescue
_ -> nil
end
package_root =
[dependency_path, Application.app_dir(:agent_blueprint_protocol), File.cwd!()]
|> Enum.reject(&is_nil/1)
|> Enum.find(fn path -> File.regular?(Path.join(path, "examples/echo-blueprint.json")) end)
true = is_binary(package_root)
blueprint_bytes = File.read!(Path.join(package_root, "examples/echo-blueprint.json"))
deployment_bytes = File.read!(Path.join(package_root, "examples/echo-deployment.json"))
blueprint_json = JSON.decode!(blueprint_bytes)
deployment_json = JSON.decode!(deployment_bytes)
{:ok, blueprint} = AgentBlueprintProtocol.decode_blueprint(blueprint_bytes)
{:ok, deployment} = AgentBlueprintProtocol.decode_deployment(deployment_bytes)
%{blueprint_decoded: true, deployment_decoded: true}%{blueprint_decoded: true, deployment_decoded: true}
Compare observed identities
Compatibility compares each manifest identity with an exact observed identity. The empty list and a matching list illustrate the two results.
Build the observed identity list from the shipped Deployment Manifest and compare both an empty and matching observation.
alias AgentBlueprintProtocol.Compatibility.Observed
support = %AgentBlueprintProtocol.Negotiation.Support{revisions: MapSet.new([1])}
{:ok, negotiation} = AgentBlueprintProtocol.negotiate(blueprint, support)
notices =
Enum.map(negotiation.notices, fn
:extension_deprecated -> :deprecated
:extension_unknown_optional_retained -> :unknown_optional
other -> {:unknown_notice, other}
end)
identities =
Enum.map(deployment_json["build_identities"], fn item ->
%{kind: item["kind"], name: item["name"], version: item["version"], digest: item["digest"]}
end)
none = AgentBlueprintProtocol.verify_compatibility(deployment, %Observed{identities: []})
full = AgentBlueprintProtocol.verify_compatibility(deployment, %Observed{identities: identities})
%{
negotiated_revision: negotiation.protocol_revision,
notices: notices,
none: case none do {:error, error} -> error.code; {:ok, _} -> :matched end,
full: match?({:ok, %AgentBlueprintProtocol.Evidence{}}, full)
}%{none: :compatibility_entry_missing, full: true, notices: [], negotiated_revision: 1}
Intersect bounds
A host can narrow an operational value. A protected narrowing returns a denial under the deny posture and a clamp record under the acknowledge posture.
Narrow the example bounds and compare the deny and acknowledge postures.
alias AgentBlueprintProtocol.BoundsAlgebra
alias AgentBlueprintProtocol.BoundsAlgebra.{BoundSet, Sources}
{:ok, blueprint_bounds} = BoundsAlgebra.from_blueprint(blueprint)
{:ok, deployment_bounds} = BoundsAlgebra.from_deployment(deployment)
{:ok, host_values} = BoundSet.to_map(deployment_bounds)
host_values = %{host_values | max_attempts: 2, disclosure_ceiling: :none}
{:ok, host_bounds} = BoundSet.new(host_values)
sources = %Sources{
blueprint: blueprint_bounds,
deployment: deployment_bounds,
host: host_bounds,
protected_clamp: :deny
}
denied = BoundsAlgebra.intersect(sources)
{:ok, acknowledged} = BoundsAlgebra.intersect(%{sources | protected_clamp: :acknowledge})
{:ok, effective} = BoundSet.to_map(acknowledged.effective)
%{
denial: case denied do {:error, error} -> error.code; {:ok, _} -> :allowed end,
effective_attempts: effective.max_attempts,
clamp_fields: Enum.map(acknowledged.clamps, & &1.field)
}%{
denial: :protected_bound_clamp_denied,
effective_attempts: 2,
clamp_fields: [:disclosure_ceiling, :max_attempts]
}
Observations
Descriptor observations can be absent, partial, or complete. The example Deployment has two tool bindings, so each map is made directly from those values.
Check none, partial, and full descriptor observations, then pass the complete set to reconcile/3.
alias AgentBlueprintProtocol.{Reconcile, BoundsAlgebra}
tool_bindings = deployment_json["tool_bindings"]
observed_all =
Map.new(tool_bindings, fn binding ->
{binding["logical_operation"], binding["descriptor_digest"]}
end)
observed_partial = Map.take(observed_all, [hd(Map.keys(observed_all))])
observation_results =
[
%AgentBlueprintProtocol.Deployment.Observations{observed: %{}},
%AgentBlueprintProtocol.Deployment.Observations{observed: observed_partial},
%AgentBlueprintProtocol.Deployment.Observations{observed: observed_all}
]
|> Enum.map(&AgentBlueprintProtocol.Deployment.verify_binding(deployment, blueprint, &1))
{:ok, host_bounds} = BoundsAlgebra.from_deployment(deployment)
support = %AgentBlueprintProtocol.Negotiation.Support{revisions: MapSet.new([1])}
inputs = %Reconcile.Inputs{
host_bounds: host_bounds,
support: support,
keys: [],
protected_clamp: :acknowledge,
observations: %AgentBlueprintProtocol.Deployment.Observations{observed: observed_all}
}
{:ok, evidence} = AgentBlueprintProtocol.reconcile(blueprint, deployment, inputs)
%{
observation_results: Enum.map(observation_results, &match?(:ok, &1)),
reconcile_succeeded: is_struct(evidence, AgentBlueprintProtocol.Evidence)
}%{observation_results: [true, true, true], reconcile_succeeded: true}
Read every evidence field
Check names and detail shapes can grow. This reader matches known shapes and keeps a catch-all. It reports the verified flag and summarizes each evidence member.
Read all Evidence fields, including verified check flags and not_verified entries, with catch-all branches for future additions.
alias AgentBlueprintProtocol.{BoundsAlgebra, Digest, Evidence}
check_view =
Enum.map(evidence.checks, fn
%{surface: surface, subject: subject, verified: verified, detail: detail} ->
detail_type =
case detail do
nil -> :none
%Evidence.Signatures{} -> :signatures
%Evidence.Binding{checked: checked, not_verified: skipped} ->
{:binding, length(checked), length(skipped)}
%BoundsAlgebra.ClampEvidence{field: field} -> {:clamp, field}
%Digest{} -> :digest
other -> {:additional_detail, is_struct(other)}
end
%{surface: surface, subject: subject, verified: verified, detail: detail_type}
other ->
%{unknown_check: is_map(other)}
end)
unverified_view =
Enum.map(evidence.not_verified, fn
:tenancy -> :host_owned
:live_policy -> :host_owned
other when is_atom(other) -> :additional_surface
_unknown -> :unknown_shape
end)
%{
protocol_revision: evidence.protocol_revision,
blueprint_digest: Digest.to_tagged(evidence.blueprint_digest),
deployment_digest: Digest.to_tagged(evidence.deployment_digest),
checks: check_view,
effective_bounds: match?({:ok, _}, BoundsAlgebra.BoundSet.to_map(evidence.effective_bounds)),
clamps: Enum.map(evidence.clamps, &{&1.field, &1.requested, &1.effective}),
optional_extensions_retained: evidence.optional_extensions_retained,
not_verified: unverified_view
}%{
protocol_revision: 1,
checks: [
%{detail: :none, subject: ["blueprint"], surface: :canonical, verified: true},
%{detail: :none, subject: ["deployment"], surface: :canonical, verified: true},
%{detail: :digest, subject: ["blueprint"], surface: :digest, verified: true},
%{detail: :digest, subject: ["deployment"], surface: :digest, verified: true},
%{detail: :none, subject: ["blueprint"], surface: :negotiation, verified: true},
%{detail: :none, subject: ["deployment"], surface: :negotiation, verified: true},
%{detail: :none, subject: ["blueprint"], surface: :structure, verified: true},
%{detail: :none, subject: ["deployment"], surface: :structure, verified: true},
%{detail: :none, subject: ["blueprint"], surface: :portability, verified: true},
%{detail: :none, subject: ["deployment"], surface: :portability, verified: true},
%{
detail: {:additional_detail, false},
subject: ["blueprint"],
surface: :signatures,
verified: false
},
%{
detail: {:additional_detail, false},
subject: ["deployment"],
surface: :signatures,
verified: false
},
%{detail: {:binding, 4, 1}, subject: ["deployment"], surface: :bind, verified: false},
%{detail: :none, subject: ["bounds"], surface: :bounds, verified: true},
%{
detail: {:clamp, :disclosure_ceiling},
subject: ["disclosure_ceiling"],
surface: :bounds,
verified: true
}
],
not_verified: [:host_owned, :host_owned, :additional_surface, :additional_surface,
:additional_surface, :additional_surface, :additional_surface, :additional_surface,
:additional_surface, :additional_surface, :additional_surface, :additional_surface],
blueprint_digest: "sha-256:b1Aw4cU5AbV9k8bdbZkRCsySDHGpTAwB-aQm57Wh7B8",
deployment_digest: "sha-256:tWFr0caS0AWFJd2UcB9gZv3kNjIUP8xZ08WWM_h8xgo",
effective_bounds: true,
clamps: [{:disclosure_ceiling, :full, :summary}],
optional_extensions_retained: []
}
Decoder limits and error shape
A lower byte limit returns the same typed Error struct used by other fallible calls.
Decode with a deliberately small byte limit and read the stable error fields.
{:error, %AgentBlueprintProtocol.Error{} = limit_error} =
AgentBlueprintProtocol.decode_blueprint(blueprint_bytes, %{bytes: 16})
%{code: limit_error.code, subject: limit_error.subject, detail: limit_error.detail}%{code: {:ceiling, :bytes}, detail: nil, subject: []}
