From 61143e85546782ad661428e47a71582657e084b2 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sun, 25 Oct 2026 03:25:30 -0400 Subject: [PATCH 1/2] fix(studio): preserve connected architecture inspection context Signed-off-by: Yordis Prieto --- docs/README.md | 2 + .../architecture-resource-roadmap.md | 18 +- docs/how-to/verify-connected-architecture.md | 113 ++ .../architecture-resource-coverage.md | 87 ++ .../source-integration-acceptance.md | 12 +- .../tests/connected_architecture_fixture.rs | 1166 +++++++++++++++++ .../src/components/ModelShell.kindmap.test.ts | 2 + .../src/components/ModelShell.kindmap.ts | 2 + .../components/ModelShell.selection.test.tsx | 17 + 9 files changed, 1410 insertions(+), 9 deletions(-) create mode 100644 docs/how-to/verify-connected-architecture.md create mode 100644 docs/reference/architecture-resource-coverage.md create mode 100644 rsworkspace/crates/trogon-atlas-server/tests/connected_architecture_fixture.rs diff --git a/docs/README.md b/docs/README.md index 05bd7f4..873d922 100644 --- a/docs/README.md +++ b/docs/README.md @@ -26,6 +26,7 @@ Solve a specific problem: - [Ship a breaking proto change](how-to/ship-a-breaking-proto-change.md) - [Start an event model from scratch](how-to/start-a-model-from-scratch.md) - [Track a version migration](how-to/track-a-migration.md) +- [Verify a connected architecture graph](how-to/verify-connected-architecture.md) ## Reference @@ -38,6 +39,7 @@ Look up exact behavior: Resource contracts and acceptance: +- [Architecture resource coverage](reference/architecture-resource-coverage.md) - [Channel and DataStore contracts](reference/channel-and-datastore-contracts.md) - [Channel and DataStore acceptance fixture](reference/channel-and-datastore-acceptance.md) - [Specification and API contracts](reference/specification-and-api-contracts.md) diff --git a/docs/explanation/architecture-resource-roadmap.md b/docs/explanation/architecture-resource-roadmap.md index 81706c4..a33b8db 100644 --- a/docs/explanation/architecture-resource-roadmap.md +++ b/docs/explanation/architecture-resource-roadmap.md @@ -1,10 +1,10 @@ # Architecture resources -Status: the native Channel and DataStore resource slice is implemented. Its -[contract reference](../reference/channel-and-datastore-contracts.md) describes -the supported fields and lifecycle. The -[acceptance fixture](../reference/channel-and-datastore-acceptance.md) links the -executable coverage and records its verification boundaries. +Status: native transport, storage, Specification, DataProduct, responsibility, +Decision and Document resources are implemented, alongside supplied-bundle +source reconciliation. The [coverage reference](../reference/architecture-resource-coverage.md) +links their field and connection contracts and separates implementation from +acceptance evidence. The remaining design scope is described below. ## Start with the behavioral model @@ -126,6 +126,14 @@ Remote connectors and additional format compilers remain separate milestones. ## Atlas Linter comes after the architecture resources +The [remaining contract boundaries](../reference/architecture-resource-coverage.md#remaining-contract-boundaries) +distinguish existing strategic references and Component membership from richer +system/service relationships, unconfirmed communication, agent metadata, +business-data relationships and arbitrary design topology. These require +explicit scope decisions. Their absence does not make the implemented native +resource slices proposals again. Connected acceptance must record which +supported relationships and client workflows were actually exercised. + Atlas Linter is deferred until the remaining architecture resources and source integration concerns are defined. Its rules can inspect model consistency, ownership, documentation, naming, contract compatibility and architectural diff --git a/docs/how-to/verify-connected-architecture.md b/docs/how-to/verify-connected-architecture.md new file mode 100644 index 0000000..dfa5f2e --- /dev/null +++ b/docs/how-to/verify-connected-architecture.md @@ -0,0 +1,113 @@ +# Verify a connected architecture graph + +Use this procedure to inspect the same disposable graph through the backend, +CLI, MCP and Studio. The [coverage reference](../reference/architecture-resource-coverage.md#composed-graph) +defines its resources and limits. Docker must be available for the test's NATS +Store. Run commands from the repository root and serialize Cargo work with +other contributors using this checkout. + +## Exercise the graph + +```sh +mise exec -- cargo test --manifest-path rsworkspace/Cargo.toml \ + -p trogon-atlas-server --test connected_architecture_fixture +``` + +This checks native selectors, reference discovery, source adoption, observation +reorder and same-store manifest reapply. The live harness is ignored during +normal testing. + +## Expose the exercised graph + +In one terminal, start the bounded manual harness: + +```sh +mise exec -- cargo test --manifest-path rsworkspace/Cargo.toml \ + -p trogon-atlas-server --test connected_architecture_fixture \ + serve_connected_architecture_fixture -- --ignored --exact --nocapture +``` + +Wait for `READY`. Retain the reported namespace and endpoints; namespace IDs +are generated. The model slug is `model`, version `1`. Default listeners are +gRPC `http://127.0.0.1:19473` and MCP `http://127.0.0.1:19474/mcp`. +Both require the synthetic test bearer `connected-fixture-admin`. This token +belongs only to the disposable fixture, not a deployment. + +`ATLAS_CONNECTED_FIXTURE_GRPC_PORT` and `ATLAS_CONNECTED_FIXTURE_MCP_PORT` +override the ports; `0` requests an available port. The harness reports the +actual endpoints. `ATLAS_CONNECTED_FIXTURE_SECONDS` sets a lifetime between +1 and 600 seconds, defaulting to 600. The deadline includes listener startup +verification. Interrupting the harness closes its listeners; the test Store +is disposable. Do not seed or restart an existing deployment for this proof. + +## Inspect with CLI and MCP + +Build the CLI, then replace `` with the READY value: + +```sh +mise exec -- cargo build --manifest-path rsworkspace/Cargo.toml \ + -p trogon-atlas-cli --bin trogon-atlas +mise exec -- rsworkspace/target/debug/trogon-atlas \ + --endpoint http://127.0.0.1:19473 --auth-token connected-fixture-admin \ + --format json describe component//receipt-service@1 +mise exec -- rsworkspace/target/debug/trogon-atlas \ + --endpoint http://127.0.0.1:19473 --auth-token connected-fixture-admin \ + --format json inspect-data-product data_product//receipts@1 +``` + +Connect an MCP client using Streamable HTTP and the same bearer. Initialize +the connection and discover `tools/list` before invoking tools. Query +`get_entity_json` for `automation_slice`, slug `observe`, exact namespace and +version `1`. Its equal ReadModel endpoints must retain distinct connection IDs. +Use `inspect_data_product` with kind `data_product`, slug `receipts` and the +same exact identity. Compare its returned ports, access selections and checked +scope with CLI inspection. + +For `list_subject_assignments`, `get_document_detail` and `get_source_run`, +`tools/list` identifies the `request_json` envelope. Its fields follow the +canonical [responsibility requests](../../proto/trogonatlas/api/eventmodel/v1alpha1/responsibility.proto), +[document requests](../../proto/trogonatlas/api/eventmodel/v1alpha1/decision_documents.proto) +and [source requests](../../proto/trogonatlas/api/eventmodel/v1alpha1/source_integration.proto). +Build requests from returned selectors, +including owner UID, local identity and digest pins. Do not guess them from +display names. The source registration/run IDs are `receipt-source` and +`receipt-title-update`; the retained outcome must be APPLIED. Product output +access must resolve independently of its declared lineage and undeclared +provider. Compare CLI/MCP response data for the same request and scope. + +Export the fixture namespace, run `diff` against that export and preview reapply +with `--dry-run`. Preserve the command results and original export. These client +checks establish export, comparison and validation-only preview. The fixture's +`assert_roundtrip` establishes persisted same-store no-op reapply. Neither proves +new-store selector rebinding or a new source publication through either client. + +## Inspect with Studio + +Build and start the repository's bridge against the fixture in another terminal: + +```sh +mise exec -- pnpm --dir tsworkspace/webapps/studio build +mise exec -- env TROGON_ATLAS_GRPC=http://127.0.0.1:19473 \ + TROGON_ATLAS_AUTH_TOKEN=connected-fixture-admin \ + TROGON_ATLAS_AUTH_PASSTHROUGH=true PORT=19475 HOST=127.0.0.1 \ + node tsworkspace/webapps/studio/server/index.mjs +``` + +Open `http://127.0.0.1:19475/em//model`, enter the fixture key when +prompted and reload the model if needed. The bridge serves the built Studio +and forwards each browser credential to the fixture backend. Stop this bridge +after the run. + +Use `?focus=data_product:/receipts`, +`?focus=decision:/delivery-choice`, +`?focus=document:/delivery-note` and +`?focus=channel:/delivery` to inspect those resources. Confirm the +receipt title is `Source receipt` and its local documentation remains intact. +Inspect exact port responsibility, accepted rationale, document content and +anchor, repeated observation selections and source run records. + +Capture screenshots, rendered text and API/console diagnostics with the exact +model identity and candidate build evidence. The fixture intentionally exposes +`SLICE_NOT_IN_STORYBOARD` and `SLICE_NO_SCENARIOS`; it is a bounded resource graph, +not a complete behavioral example. The stock Studio e2e suite uses a fake gRPC +backend and does not substitute for this real-store inspection. diff --git a/docs/reference/architecture-resource-coverage.md b/docs/reference/architecture-resource-coverage.md new file mode 100644 index 0000000..95a178f --- /dev/null +++ b/docs/reference/architecture-resource-coverage.md @@ -0,0 +1,87 @@ +# Architecture resource coverage + +This reference separates native resource support from connection coverage and +acceptance evidence. A retained field or artifact does not establish a checked +relationship, executable behavior, or a passing client workflow. See the +[roadmap](../explanation/architecture-resource-roadmap.md) for design boundaries. + +## Native resource contracts + +| Resource | Supported fields and selections | Contract and executable coverage | +| --- | --- | --- | +| Strategic scope | Project, Domain, Subdomain and BoundedContext records; project/domain references, realized subdomains and directed context agreements. | [Strategic schema](../../proto/trogonatlas/eventmodel/v1alpha1/strategic.proto), [reference extraction](../../rsworkspace/crates/trogon-atlas-core/src/refs.rs). | +| Component and ExternalSystem | Component technology and curated exact entity membership; external boundary descriptions, ReadModel external-source links and Processor calls. | [Infrastructure schema](../../proto/trogonatlas/eventmodel/v1alpha1/infrastructure.proto), [reference extraction](../../rsworkspace/crates/trogon-atlas-core/src/refs.rs). | +| Channel | Address parameters, protocols, delivery declarations, optional broker and bindings to exact AutomationSlice observation occurrences. | [Contract](channel-and-datastore-contracts.md), [acceptance](channel-and-datastore-acceptance.md). | +| DataStore | Store class, technology, purpose, authority, access mode, classification, residency, retention and identified ReadModel attachments. | [Contract](channel-and-datastore-contracts.md), [acceptance](channel-and-datastore-acceptance.md). | +| Specification | Retained artifacts, schema/message/operation members, individual message mappings, operation roles, provider exposures and usages. | [Contract](specification-and-api-contracts.md), [acceptance](specification-and-api-acceptance.md). | +| DataProduct | Input/output ports, output contracts, selected storage/channel/operation access and declared lineage; inspection reports checked model paths separately. | [Contract](data-product-contracts.md), [acceptance](data-product-acceptance.md). | +| Team and Person | Directory records, identified membership and explicit responsibility for an exact entity, supported connection occurrence, Specification member or DataProduct port. | [Contract](responsibility-contracts.md), [acceptance](responsibility-acceptance.md). | +| Decision | Subjects, disposition, options, selected option, rationale, consequences, replacement and supporting Document selections. | [Contract](decision-and-document-contracts.md), [acceptance](decision-and-document-acceptance.md). | +| Document | Subjects, retained content or link, content digest, identified anchors and navigation. | [Contract](decision-and-document-contracts.md), [acceptance](decision-and-document-acceptance.md). | +| Source control | Registrations, immutable supplied snapshots, explicit mappings/adoption, field ownership, overlays, guarded runs and durable publication/recovery. Source records are separate from native entity kinds. | [Contract](source-integration-contracts.md), [API](source-control-api.md), [acceptance](source-integration-acceptance.md). | + +Shared native identity uses exact kind, namespace, slug and numeric version. +Contained selections additionally use the returned owner incarnation and local +identity required by their contract. Titles, names, list positions and source +version strings do not replace these selectors. See +[connection identity](connection-identity.md). + +## Remaining contract boundaries + +| Area | Existing representation | Remaining boundary | +| --- | --- | --- | +| Strategic membership | Domain project, Subdomain domain and context realizations are native references. | Arbitrary domain nesting and shared authored system/service membership have no dedicated fields. Domain listing derives membership from namespace and one realized subdomain; its maps omit versions. It does not prove exact-version or multi-domain membership. | +| System and service relationships | Component members can select other Components and ExternalSystems, with ordinary reference discovery. | Dedicated grouping levels, labeled directed system relationships and system-level actor interactions require a contract decision. Containment is Component-owned; ExternalSystem has no members field, so external-system-owned membership also remains a contract decision. | +| Communication | API roles/providers/usages, Processor calls and Channel bindings to actual observations are supported. | General authored send/receive and conditional-trigger declarations before behavioral mapping have no native relationship contract. Retained source declarations do not establish executable paths. | +| Metadata and agents | Registered annotations include links, code repository/path/symbol/ref provenance, tags and lifecycle status. | Lifecycle effective dates and agent model/tool declarations need dedicated contracts. Arbitrary metadata references are not recursively indexed. See [annotations](../../proto/trogonatlas/annotation/v1alpha1/annotations.proto). | +| Business data | Schema fields, nested shapes, schema type references and identity-derivation annotations are available. | Business identifier roles, foreign entity/property selections, relationship labels and general aggregate intent are not supplied by type reuse. See [field schema](../../proto/trogonatlas/eventmodel/v1alpha1/fields.proto). | +| Design graphs | Storyboards retain ordered slices and traces; Documents retain content, anchors and structured navigation. | Generic branches, loops and subflows remain document content without typed topology traversal. Parsing diagrams and validating executable examples require separate contracts. | +| Dependency explanations | References expose actual field paths; impact follows graph reachability; DataProduct inspection checks selected access. | These queries do not classify every runtime service dependency or recover relationships absent from native contracts. | + +These boundaries are scoped design decisions, not implicit requirements to add +resource kinds. Supplied-bundle source reconciliation is implemented; remote +connectors, automatic synchronization, additional semantic format compilers, +portable selector rebinding and the WASM/WIT linter host remain separate work. + +## Evidence boundaries + +The linked acceptance references identify separate executable fixtures and +remaining requirements. The resource-specific fixtures use independent stores +and namespaces. The composed fixture below supplies an additional shared graph; +neither its existence nor separate passing tests prove every client workflow. + +Backend service tests, authenticated transport tests, CLI/MCP tests, Studio +component tests and browser runs establish different boundaries. Browser +evidence must identify its backend and fixture; a fake gRPC backend cannot +establish native persistence or source publication. A native manifest roundtrip +in the same store does not establish portable migration of incarnation-bound +selectors into a new store. + +## Composed graph + +The [composed fixture](../../rsworkspace/crates/trogon-atlas-server/tests/connected_architecture_fixture.rs) +creates a registered namespace and retains the actual returned owner UIDs, +connection IDs, member IDs, port IDs and anchor IDs. Its logical identities use +version `1`; the generated namespace is reported by the live harness. + +| Entry point | Connected declarations and checks | +| --- | --- | +| Component `receipt-service` | Exact membership of Processor `notifier`, ReadModel `receipt`, Specification `api` and DataProduct `receipts`. | +| AutomationSlice `observe` | Repeated observations of `receipt` have distinct identities. Channel `delivery` binds both; DataStore `warehouse` attaches to the same ReadModel. Reordering observations preserves their selectors. | +| Specification `api` | A response operation selects the returned Receipt message member mapped to `receipt`. Product output access selects that exact response, storage attachment and both channel observations. An undeclared provider remains undeclared. | +| Product `receipts` | Input/output ports retain identity. Declared lineage is reported separately from resolved access and data selections. | +| Person `maintainer` and Team `receipt-team` | Exact observation/member/port assignments and Team membership remain queryable. Assignment to one observation does not spread to its equal-endpoint sibling. | +| Decision `delivery-choice` and Document `delivery-note` | Accepted rationale, exact subjects, retained Markdown and a digest-pinned anchor remain connected through typed queries and reverse discovery. | +| Source registration `receipt-source` | A PARTIAL immutable bundle and explicit adoption own only the receipt title. Preview leaves native state unchanged; apply and replay preserve local documentation, all other owners and retained original bytes. | + +The fixture asserts expected outgoing endpoint multiplicities, unique field +paths and matching incoming references. Native manifest export/reapply in the +same store preserves content, etags and returned identities. It does not assert +portable rebinding, all strategic-scope cases, every resource lifecycle case, +or the broader source capture matrix. + +The ignored `serve_connected_architecture_fixture` test exposes that same +exercised graph over authenticated loopback gRPC and MCP for a bounded manual +run. See [connected acceptance](../how-to/verify-connected-architecture.md) for +CLI, MCP and Studio verification. Normal test execution does not start these +manual listeners. diff --git a/docs/reference/source-integration-acceptance.md b/docs/reference/source-integration-acceptance.md index 8feaffd..fc8e3ea 100644 --- a/docs/reference/source-integration-acceptance.md +++ b/docs/reference/source-integration-acceptance.md @@ -16,14 +16,17 @@ They cover intake only. Source-specific evidence is separated below. | --- | --- | | [Core ownership and merge](../../rsworkspace/crates/trogon-atlas-core/tests/source_integration.rs) | Closed ownership units, presence, ordered values, merge dispositions and profile bounds. | | [Connected fixture](../../rsworkspace/crates/trogon-atlas-server/tests/source_integration_fixture.rs) | Explicit adoption, repeated native observations, immutable bytes, ordered overlay fidelity, separate accepted source/native bases, drift and replay. | +| [Composed architecture fixture](../../rsworkspace/crates/trogon-atlas-server/tests/connected_architecture_fixture.rs) | One real-store graph with Component membership, Channel/DataStore, Specification, DataProduct, responsibility and Decision/Document selectors; title-only source adoption preserves other native owners and exact selections through reorder and same-store manifest reapply. | | [Backend boundaries](../../rsworkspace/crates/trogon-atlas-server/tests/source_integration_backend.rs) | Authenticated scope and pagination, incomplete payload fidelity, independent retention, quotas, prepare interruption, tenant types and phase deadlines. | | [Lifecycle and selectors](../../rsworkspace/crates/trogon-atlas-server/tests/source_lifecycle_acceptance.rs) | Creation, stale guards, partial/complete inventory, safe deletion, final overlay, branches, cancellation and exact descriptive local selectors. | | [Store publication](../../rsworkspace/crates/trogon-atlas-store/tests/source_publication.rs) | Baseline/branch fault boundaries, restarted writer epochs, reader fencing, exact postimages, chunk reclamation and physical quota accounting. | | [Native client workflow](../../rsworkspace/crates/trogon-atlas-client/tests/source_backend_roundtrip.rs), [CLI](../../rsworkspace/crates/trogon-atlas-cli/tests/source_cli_integration.rs), [MCP](../../rsworkspace/crates/trogon-atlas-mcp/tests/source_mcp_integration.rs) | Authenticated registration, snapshot, mapping, preview, apply and report workflows, including tenant type context and native-owner input. | A test's presence is not a passing run or proof of every requirement below. -The independent fixture composition described below remains a broader capture -scenario. Existing registry rules do not permit a read-scoped Admin, while +The composed architecture fixture provides a bounded shared graph. The complete +independent-fixture capture described below remains a broader scenario; +the composed graph does not cover every table row. Existing registry rules do +not permit a read-scoped Admin, while verified COMPLETE inventory is Admin-only. Source deletion tests therefore separate scoped Writer permission refusal from an Admin's surviving-referrer guard; they do not claim a hidden-to-applying-principal deletion execution. @@ -31,8 +34,9 @@ guard; they do not claim a hidden-to-applying-principal deletion execution. ## Connected fixture inputs Capture a bounded native manifest/file bundle from the existing fixtures, keeping -each fixture's namespaces separate. Fixture composition is a future acceptance -extension, not an assertion that these independent tests share a running store. +each fixture's namespaces separate. This full capture remains an acceptance +extension beyond the bounded [composed graph](architecture-resource-coverage.md#composed-graph). +The independent tests below do not share a running store. All identities below have numeric version 1. Record exact kind and namespace; display names never substitute for returned local IDs or owner UIDs. diff --git a/rsworkspace/crates/trogon-atlas-server/tests/connected_architecture_fixture.rs b/rsworkspace/crates/trogon-atlas-server/tests/connected_architecture_fixture.rs new file mode 100644 index 0000000..7492ec1 --- /dev/null +++ b/rsworkspace/crates/trogon-atlas-server/tests/connected_architecture_fixture.rs @@ -0,0 +1,1166 @@ +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use std::{ + collections::{HashMap, HashSet}, + fmt::Write as _, + sync::{atomic::AtomicI64, Arc}, + time::Duration, +}; + +use pb::event_model_service_server::EventModelService; +use serde_json::json; +use sha2::{Digest, Sha256}; +use tonic::Request; +use trogon_atlas_proto as pb; +use trogon_atlas_server::{ + auth::{BearerAuth, TokenRegistry}, + mcp::{data_plane, mcp_router}, + metrics::ServerMetrics, + service::EventModelServiceImpl, + telemetry::{InflightLayer, RpcMetricsLayer}, +}; + +struct Fixture { + service: EventModelServiceImpl, + namespace: String, + expected: HashMap, +} + +fn reference(entity: &pb::Entity) -> pb::EntityRef { + trogon_atlas_server::conv::entity_to_ref(entity).unwrap() +} + +fn uid(entity: &pb::Entity) -> String { + entity.system.as_ref().unwrap().uid.clone() +} + +fn connection(entity: &pb::Entity, id: Option) -> pb::ConnectionRef { + assert!(id.is_some()); + pb::ConnectionRef { + owner: Some(reference(entity)), + owner_uid: uid(entity), + id, + } +} + +fn name(value: &str) -> pb::MemberName { + pb::MemberName { + value: value.into(), + } +} + +fn title(value: &str) -> pb::CatalogTitle { + pb::CatalogTitle { + value: value.into(), + } +} + +fn text(value: &str) -> pb::CatalogText { + pb::CatalogText { + value: value.into(), + } +} + +fn associations(subjects: &[pb::CatalogSubject]) -> Vec { + subjects + .iter() + .map(|subject| pb::SubjectAssociation { + subject: Some(subject.clone()), + ..Default::default() + }) + .collect() +} + +impl Fixture { + fn id(&self, slug: &str) -> pb::Id { + pb::Id { + namespace: self.namespace.clone(), + slug: slug.into(), + version: 1, + } + } + + fn model(&self) -> pb::ReadModelRef { + pb::ReadModelRef { + id: Some(self.id("receipt")), + } + } + + fn remember(&mut self, entity: &pb::Entity) { + let owner = reference(entity); + self.expected.insert( + pb::EntityKey::new( + pb::EntityKind::try_from(owner.kind).unwrap(), + owner.id.as_ref().unwrap(), + ), + entity.clone(), + ); + } + + async fn put(&mut self, kind: pb::entity::Kind) -> pb::Entity { + let proposed = pb::Entity { + kind: Some(kind), + ..Default::default() + }; + let owner = reference(&proposed); + let key = pb::EntityKey::new( + pb::EntityKind::try_from(owner.kind).unwrap(), + owner.id.as_ref().unwrap(), + ); + let if_match = if self.expected.contains_key(&key) { + self.get(&proposed).await.etag + } else { + String::new() + }; + let response = self + .service + .put_entity(Request::new(pb::PutEntityRequest { + entity: Some(proposed), + if_match, + ..Default::default() + })) + .await + .unwrap_or_else(|error| panic!("fixture put failed for {owner:?}: {error}")) + .into_inner(); + let entity = response.entity.unwrap(); + self.remember(&entity); + entity + } + + async fn get(&self, entity: &pb::Entity) -> pb::GetEntityResponse { + let owner = reference(entity); + self.service + .get_entity(Request::new(pb::GetEntityRequest { + kind: owner.kind, + id: owner.id, + })) + .await + .unwrap() + .into_inner() + } + + async fn assert_bidirectional_references(&self, required: &[(&pb::Entity, Vec<&pb::Entity>)]) { + assert_eq!(required.len(), self.expected.len()); + for (entity, targets) in required { + let owner = reference(entity); + let outgoing = self + .service + .get_outgoing_references(Request::new(pb::GetReferencesRequest { + kind: owner.kind, + id: owner.id.clone(), + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + assert_eq!(outgoing.next_page_token, ""); + assert_eq!( + outgoing.references.len(), + targets.len(), + "reference count for {owner:?}" + ); + assert_eq!( + outgoing + .references + .iter() + .map(|edge| &edge.from_field) + .collect::>() + .len(), + targets.len() + ); + for target in targets { + let target = reference(target); + assert_eq!( + outgoing + .references + .iter() + .filter(|edge| edge.to.as_ref() == Some(&target)) + .count(), + targets + .iter() + .filter(|expected| reference(expected) == target) + .count(), + "missing authored endpoint {owner:?} -> {target:?}" + ); + } + for edge in &outgoing.references { + assert_eq!(edge.from.as_ref(), Some(&owner)); + assert_eq!(edge.from_uid, uid(entity)); + let target = edge.to.as_ref().unwrap(); + let incoming = self + .service + .get_incoming_references(Request::new(pb::GetReferencesRequest { + kind: target.kind, + id: target.id.clone(), + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + assert_eq!(incoming.next_page_token, ""); + assert!( + incoming.references.contains(edge), + "missing reverse reference {edge:?}" + ); + } + } + } + + async fn manifests(&self) -> String { + let exported = self + .service + .export_event_model_manifests(Request::new(pb::ExportEventModelManifestsRequest { + id: Some(self.id("model")), + })) + .await + .unwrap() + .into_inner(); + assert!(!exported.truncated); + assert_eq!(exported.skipped, Vec::::new()); + assert_eq!(exported.external_manifests, ""); + exported.manifests + } + + async fn assert_roundtrip(&self) { + let manifests = trogon_atlas_client::manifest::parse_str( + &self.manifests().await, + "connected-architecture-export", + ) + .unwrap(); + assert_eq!(manifests.len(), self.expected.len()); + for manifest in manifests { + assert_eq!(manifest.id.namespace, self.namespace); + assert_eq!(manifest.id.version, 1); + let original = self + .expected + .get(&pb::EntityKey::new(manifest.kind, &manifest.id)) + .unwrap(); + assert_eq!(manifest.entity.kind, original.kind); + let before = self.get(original).await; + let response = self + .service + .put_entity(Request::new(pb::PutEntityRequest { + entity: Some(manifest.entity), + if_match: before.etag.clone(), + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + assert!(response.no_op); + assert_eq!(response.etag, before.etag); + assert_eq!(response.entity.as_ref(), Some(original)); + } + } +} + +#[tokio::test] +async fn composed_architecture_preserves_exact_selectors_through_source_apply_and_roundtrip() { + Box::pin(exercise()).await; +} + +async fn exercise() -> Fixture { + let service = EventModelServiceImpl::try_new( + Arc::new(trogon_atlas_testsupport::shared().await.store().await), + ServerMetrics::default(), + ) + .unwrap(); + let namespace = service + .register_namespace(Request::new(pb::RegisterNamespaceRequest { + name: "connected-architecture".into(), + parent: "fixture-owner".into(), + })) + .await + .unwrap() + .into_inner() + .namespace + .unwrap() + .id; + let mut fixture = Fixture { + service, + namespace, + expected: HashMap::new(), + }; + let receipt = fixture + .put(pb::entity::Kind::ReadModel(pb::ReadModel { + id: Some(fixture.id("receipt")), + title: "Receipt".into(), + doc: "Locally maintained receipt semantics".into(), + ..Default::default() + })) + .await; + let notifier = fixture + .put(pb::entity::Kind::Processor(pb::Processor { + id: Some(fixture.id("notifier")), + title: "Notifier".into(), + ..Default::default() + })) + .await; + let observed = fixture + .put(pb::entity::Kind::AutomationSlice(pb::AutomationSlice { + id: Some(fixture.id("observe")), + title: "Observe receipts".into(), + source_read_models: vec![ + pb::ReadModelEdge { + read_model: Some(fixture.model()), + ..Default::default() + }; + 2 + ], + processor: Some(pb::ProcessorEdge { + processor: Some(pb::ProcessorRef { + id: Some(fixture.id("notifier")), + }), + ..Default::default() + }), + ..Default::default() + })) + .await; + let Some(pb::entity::Kind::AutomationSlice(slice)) = &observed.kind else { + panic!("slice") + }; + let observations: Vec<_> = slice + .source_read_models + .iter() + .map(|edge| connection(&observed, edge.connection_id.clone())) + .collect(); + assert_ne!(observations[0], observations[1]); + let channel = fixture + .put(pb::entity::Kind::Channel(pb::Channel { + id: Some(fixture.id("delivery")), + title: "Delivery".into(), + bindings: observations + .iter() + .map(|observation| pb::ChannelBinding { + observation: Some(observation.clone()), + ..Default::default() + }) + .collect(), + ..Default::default() + })) + .await; + let store = fixture + .put(pb::entity::Kind::DataStore(pb::DataStore { + id: Some(fixture.id("warehouse")), + title: "Warehouse".into(), + store_class: Some(pb::StoreClass { + kind: Some(pb::store_class::Kind::Other("warehouse".into())), + }), + attachments: vec![pb::ReadModelAttachment { + read_model: Some(fixture.model()), + ..Default::default() + }], + ..Default::default() + })) + .await; + let specification = fixture + .put(pb::entity::Kind::Specification(pb::Specification { + id: Some(fixture.id("api")), + title: "Receipt API".into(), + messages: vec![pb::ContractMessage { + name: Some(name("Receipt")), + mapping: Some(pb::MessageReadModelMapping { + read_model: Some(fixture.model()), + ..Default::default() + }), + ..Default::default() + }], + ..Default::default() + })) + .await; + let Some(pb::entity::Kind::Specification(mut spec)) = specification.kind.clone() else { + panic!("specification") + }; + spec.operations.push(pb::ContractOperation { + name: Some(name("ReadReceipt")), + messages: vec![pb::OperationMessage { + message_id: spec.messages[0].member_id.clone(), + role: pb::operation_message::Role::Response as i32, + ..Default::default() + }], + ..Default::default() + }); + let specification = fixture.put(pb::entity::Kind::Specification(spec)).await; + let Some(pb::entity::Kind::Specification(spec)) = &specification.kind else { + panic!("specification") + }; + let member = pb::ContractMemberRef { + specification: Some(pb::SpecificationRef { + id: Some(fixture.id("api")), + }), + owner_uid: uid(&specification), + kind: pb::ContractMemberKind::Message as i32, + member_id: spec.messages[0].member_id.clone(), + expected_digest: None, + }; + let Some(pb::entity::Kind::DataStore(storage)) = &store.kind else { + panic!("store") + }; + let Some(pb::entity::Kind::Channel(transport)) = &channel.kind else { + panic!("channel") + }; + assert_ne!( + transport.bindings[0].connection_id, + transport.bindings[1].connection_id + ); + let mut access = vec![pb::OutputAccess { + selection: Some(pb::output_access::Selection::Store(pb::StoreAccess { + store: Some(pb::DataStoreRef { + id: Some(fixture.id("warehouse")), + }), + attachment: Some(connection( + &store, + storage.attachments[0].connection_id.clone(), + )), + })), + ..Default::default() + }]; + access.extend( + transport + .bindings + .iter() + .zip(&observations) + .map(|(binding, observation)| pb::OutputAccess { + selection: Some(pb::output_access::Selection::Channel(pb::ChannelAccess { + channel: Some(pb::ChannelRef { + id: Some(fixture.id("delivery")), + }), + binding: Some(connection(&channel, binding.connection_id.clone())), + observation: Some(observation.clone()), + })), + ..Default::default() + }), + ); + access.push(pb::OutputAccess { + selection: Some(pb::output_access::Selection::Operation( + pb::OperationAccess { + operation: Some(pb::ContractMemberRef { + kind: pb::ContractMemberKind::Operation as i32, + member_id: spec.operations[0].member_id.clone(), + ..member.clone() + }), + response: Some(connection( + &specification, + spec.operations[0].messages[0].connection_id.clone(), + )), + provider_exposure: None, + }, + )), + ..Default::default() + }); + let product = fixture + .put(pb::entity::Kind::DataProduct(pb::DataProduct { + id: Some(fixture.id("receipts")), + title: "Receipt data".into(), + inputs: vec![pb::DataInputPort { + name: Some(pb::PortName { + value: "Observed receipts".into(), + }), + data: Some(pb::InputDataSelection { + target: Some(pb::input_data_selection::Target::ReadModel(fixture.model())), + ..Default::default() + }), + ..Default::default() + }], + outputs: vec![pb::DataOutputPort { + name: Some(pb::PortName { + value: "Receipts".into(), + }), + data: Some(pb::OutputDataSelection { + read_model: Some(fixture.model()), + ..Default::default() + }), + access, + ..Default::default() + }], + ..Default::default() + })) + .await; + let Some(pb::entity::Kind::DataProduct(mut data)) = product.kind.clone() else { + panic!("product") + }; + data.outputs[0].derived_from = Some(pb::DeclaredLineage { + inputs: vec![pb::LineageInput { + input_port_id: data.inputs[0].port_id.clone(), + ..Default::default() + }], + ..Default::default() + }); + let product = fixture.put(pb::entity::Kind::DataProduct(data)).await; + let Some(pb::entity::Kind::DataProduct(data)) = &product.kind else { + panic!("product") + }; + let port = pb::DataPortRef { + product: Some(pb::DataProductRef { + id: Some(fixture.id("receipts")), + }), + owner_uid: uid(&product), + direction: pb::DataPortDirection::Output as i32, + port_id: data.outputs[0].port_id.clone(), + }; + let responsibilities = vec![ + pb::responsibility_subject::Selection::Connection(observations[0].clone()), + pb::responsibility_subject::Selection::Member(member.clone()), + pb::responsibility_subject::Selection::Port(port.clone()), + ] + .into_iter() + .map(|selection| pb::Responsibility { + kind: pb::ResponsibilityKind::Owner as i32, + subject: Some(pb::ResponsibilitySubject { + selection: Some(selection), + }), + ..Default::default() + }) + .collect(); + let person = fixture + .put(pb::entity::Kind::Person(pb::Person { + id: Some(fixture.id("maintainer")), + title: Some(pb::DirectoryDisplayName { + value: "Receipt maintainer".into(), + }), + responsibilities, + ..Default::default() + })) + .await; + let subjects: Vec<_> = [ + pb::catalog_subject::Selection::Connection(observations[0].clone()), + pb::catalog_subject::Selection::Connection(observations[1].clone()), + pb::catalog_subject::Selection::Member(member), + pb::catalog_subject::Selection::Port(port), + ] + .into_iter() + .map(|selection| pb::CatalogSubject { + selection: Some(selection), + }) + .collect(); + let content = "# Receipt delivery\nDelivery remains a runtime responsibility.\n"; + let mut sha256 = String::with_capacity(64); + for byte in Sha256::digest(content.as_bytes()) { + write!(sha256, "{byte:02x}").unwrap(); + } + let digest = pb::ContentDigest { sha256 }; + let document = fixture + .put(pb::entity::Kind::Document(pb::Document { + id: Some(fixture.id("delivery-note")), + title: Some(title("Delivery note")), + subjects: associations(&subjects), + body: Some(pb::document::Body::Retained(pb::RetainedDocumentText { + media_type: Some(pb::MediaType { + value: "text/markdown".into(), + }), + content: content.into(), + digest: Some(digest.clone()), + })), + anchors: vec![pb::DocumentAnchor { + fragment: Some(pb::AnchorFragment { + value: "delivery".into(), + }), + label: Some(title("Delivery")), + ..Default::default() + }], + ..Default::default() + })) + .await; + let Some(pb::entity::Kind::Document(doc)) = &document.kind else { + panic!("document") + }; + let pinned = pb::DocumentSelection { + selection: Some(pb::document_selection::Selection::Anchor( + pb::DocumentAnchorRef { + document: Some(pb::DocumentRef { + id: Some(fixture.id("delivery-note")), + }), + owner_uid: uid(&document), + anchor_id: doc.anchors[0].anchor_id.clone(), + expected_digest: Some(digest), + }, + )), + }; + let decision = fixture + .put(pb::entity::Kind::Decision(pb::Decision { + id: Some(fixture.id("delivery-choice")), + title: Some(title("Receipt delivery")), + subjects: associations(&subjects), + status: pb::DecisionStatus::Proposed as i32, + context: Some(text("Choose an explicit observation.")), + rationale: Some(text("Keep both repeated observations distinct.")), + options: vec![pb::DecisionOption { + title: Some(title("Declared delivery")), + description: Some(text("Use the explicitly selected observation.")), + ..Default::default() + }], + documents: vec![pb::SupportingDocument { + document: Some(pinned.clone()), + ..Default::default() + }], + ..Default::default() + })) + .await; + let Some(pb::entity::Kind::Decision(mut choice)) = decision.kind else { + panic!("decision") + }; + choice.status = pb::DecisionStatus::Accepted as i32; + choice.selected_option = choice.options[0].option_id.clone(); + choice.consequences.push(pb::DecisionConsequence { + kind: pb::decision_consequence::Kind::Risk as i32, + text: Some(text("Delivery still requires runtime processing.")), + }); + let decision = fixture.put(pb::entity::Kind::Decision(choice)).await; + let team = fixture + .put(pb::entity::Kind::Team(pb::Team { + id: Some(fixture.id("receipt-team")), + title: Some(pb::DirectoryDisplayName { + value: "Receipt team".into(), + }), + members: vec![pb::TeamMember { + person: Some(pb::PersonRef { + id: Some(fixture.id("maintainer")), + }), + ..Default::default() + }], + responsibilities: [&product, &decision, &document] + .into_iter() + .map(|entity| pb::Responsibility { + kind: pb::ResponsibilityKind::Steward as i32, + subject: Some(pb::ResponsibilitySubject { + selection: Some(pb::responsibility_subject::Selection::Entity(reference( + entity, + ))), + }), + ..Default::default() + }) + .collect(), + ..Default::default() + })) + .await; + let component = fixture + .put(pb::entity::Kind::Component(pb::Component { + id: Some(fixture.id("receipt-service")), + title: "Receipt service".into(), + members: [¬ifier, &receipt, &specification, &product] + .into_iter() + .map(reference) + .collect(), + ..Default::default() + })) + .await; + let model = fixture + .put(pb::entity::Kind::EventModel(pb::EventModel { + id: Some(fixture.id("model")), + title: "Connected receipt architecture".into(), + members: fixture.expected.values().map(reference).collect(), + ..Default::default() + })) + .await; + + let required = [ + (&receipt, vec![]), + ( + &component, + vec![¬ifier, &receipt, &specification, &product], + ), + (¬ifier, vec![]), + (&observed, vec![&receipt, &receipt, ¬ifier]), + (&channel, vec![&observed, &observed]), + (&store, vec![&receipt]), + (&specification, vec![&receipt]), + ( + &product, + vec![ + &receipt, + &receipt, + &store, + &store, + &channel, + &channel, + &channel, + &channel, + &observed, + &observed, + &specification, + &specification, + ], + ), + (&person, vec![&observed, &specification, &product]), + ( + &document, + vec![&observed, &observed, &specification, &product], + ), + ( + &decision, + vec![&observed, &observed, &specification, &product, &document], + ), + (&team, vec![&person, &product, &decision, &document]), + ( + &model, + vec![ + &receipt, + ¬ifier, + &observed, + &channel, + &store, + &specification, + &product, + &person, + &document, + &decision, + &team, + &component, + ], + ), + ]; + fixture.assert_bidirectional_references(&required).await; + fixture.assert_roundtrip().await; + adopt_receipt_title(&mut fixture, &receipt).await; + let Some(pb::entity::Kind::AutomationSlice(mut reordered)) = observed.kind.clone() else { + panic!("slice") + }; + reordered.source_read_models.reverse(); + let reordered = fixture + .put(pb::entity::Kind::AutomationSlice(reordered)) + .await; + assert_eq!(uid(&reordered), uid(&observed)); + let Some(pb::entity::Kind::AutomationSlice(reordered)) = reordered.kind else { + panic!("slice") + }; + assert_eq!(reordered.source_read_models[0], slice.source_read_models[1]); + assert_eq!(reordered.source_read_models[1], slice.source_read_models[0]); + let inspection = fixture + .service + .inspect_data_product(Request::new(pb::InspectDataProductRequest { + product: Some(pb::DataProductRef { + id: Some(fixture.id("receipts")), + }), + })) + .await + .unwrap() + .into_inner(); + assert!(!inspection.incomplete); + assert_eq!(inspection.product.as_ref(), Some(data)); + let output = inspection + .ports + .iter() + .find(|state| state.port.as_ref().unwrap().port_id == data.outputs[0].port_id) + .unwrap(); + assert_eq!(output.access.len(), data.outputs[0].access.len()); + assert!(output + .access + .iter() + .all(|access| access.state == pb::DataProductSelectionState::Resolved as i32)); + assert_eq!( + output.access.last().unwrap().provider, + pb::DataProductSelectionState::Undeclared as i32 + ); + for subject in &subjects { + let found = fixture + .service + .list_subject_decisions(Request::new(pb::ListSubjectDecisionsRequest { + subject: Some(subject.clone()), + status: Some(pb::DecisionStatus::Accepted as i32), + })) + .await + .unwrap() + .into_inner(); + assert!(!found.incomplete); + assert_eq!( + found.subject_state, + pb::CatalogResolutionState::Resolved as i32 + ); + assert_eq!(found.declarations.len(), 1); + assert_eq!( + found.declarations[0].owner.as_ref().unwrap().uid, + uid(&decision) + ); + let docs = fixture + .service + .list_subject_documents(Request::new(pb::ListSubjectDocumentsRequest { + subject: Some(subject.clone()), + })) + .await + .unwrap() + .into_inner(); + assert!(!docs.incomplete); + assert_eq!( + docs.subject_state, + pb::CatalogResolutionState::Resolved as i32 + ); + assert_eq!(docs.declarations.len(), 1); + assert_eq!( + docs.declarations[0].owner.as_ref().unwrap().uid, + uid(&document) + ); + } + let Some(pb::entity::Kind::Person(maintainer)) = &person.kind else { + panic!("person") + }; + for assignment in &maintainer.responsibilities { + let found = fixture + .service + .list_subject_assignments(Request::new(pb::ListSubjectAssignmentsRequest { + subject: assignment.subject.clone(), + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + assert!(!found.incomplete); + assert_eq!( + found.subject_state, + pb::ResponsibilityResolutionState::Resolved as i32 + ); + assert_eq!(found.assignments.len(), 1); + assert_eq!(found.assignments[0].assignee_uid, uid(&person)); + assert_eq!(found.assignments[0].assignment.as_ref(), Some(assignment)); + } + let sibling = fixture + .service + .list_subject_assignments(Request::new(pb::ListSubjectAssignmentsRequest { + subject: Some(pb::ResponsibilitySubject { + selection: Some(pb::responsibility_subject::Selection::Connection( + observations[1].clone(), + )), + }), + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + assert!(!sibling.incomplete); + assert_eq!( + sibling.assignments, + Vec::::new() + ); + let memberships = fixture + .service + .list_person_teams(Request::new(pb::ListPersonTeamsRequest { + person: Some(pb::PersonRef { + id: Some(fixture.id("maintainer")), + }), + })) + .await + .unwrap() + .into_inner(); + let Some(pb::entity::Kind::Team(team_value)) = &team.kind else { + panic!("team") + }; + assert_eq!(memberships.memberships.len(), 1); + assert_eq!(memberships.memberships[0].team_uid, uid(&team)); + assert_eq!( + memberships.memberships[0].membership.as_ref(), + Some(&team_value.members[0]) + ); + let detail = fixture + .service + .get_document_detail(Request::new(pb::GetDocumentDetailRequest { + document: Some(pinned.clone()), + })) + .await + .unwrap() + .into_inner(); + assert_eq!( + detail.selection_state, + pb::CatalogResolutionState::Resolved as i32 + ); + assert_eq!(detail.anchor.as_ref(), doc.anchors.first()); + let incoming = fixture + .service + .list_incoming_document_declarations(Request::new( + pb::ListIncomingDocumentDeclarationsRequest { + document: Some(pinned), + }, + )) + .await + .unwrap() + .into_inner(); + assert!(!incoming.incomplete); + assert_eq!(incoming.declarations.len(), 1); + assert_eq!( + incoming.declarations[0].owner.as_ref().unwrap().uid, + uid(&decision) + ); + fixture.assert_bidirectional_references(&required).await; + fixture.assert_roundtrip().await; + fixture +} + +async fn adopt_receipt_title(fixture: &mut Fixture, receipt: &pb::Entity) { + let original = fixture.get(receipt).await; + let registration = fixture + .service + .create_source_registration(Request::new(pb::CreateSourceRegistrationRequest { + registration: Some(pb::SourceRegistration { + id: "receipt-source".into(), + administration_namespace: fixture.namespace.clone(), + destination_namespaces: vec![fixture.namespace.clone()], + profile_revision: 1, + ownership_policy_revision: "title-only".into(), + deletion_policy: pb::SourceDeletionPolicy::RetainNative as i32, + ..Default::default() + }), + })) + .await + .unwrap() + .into_inner() + .registration + .unwrap(); + let inventory = serde_json::to_vec(&json!({ + "profile": "supplied-native-v1", "profile_revision": 1, + "registration": registration.id, "source_version": "00.RC-A", + "partition": {"categories": ["model"], "logical_roots": ["native"], "selected_keys": []}, + "coverage": "PARTIAL", + "resources": [{"key": {"category": "model", "key": "receipt"}, "file": "native/model.yaml", "pointer": {"kind": "native", "raw": "#/receipt"}}], + "relations": [] + })).unwrap(); + let bytes = fixture.manifests().await.into_bytes(); + let snapshot = fixture + .service + .accept_source_snapshot(Request::new(pb::AcceptSourceSnapshotRequest { + registration_id: registration.id.clone(), + registration_revision: registration.revision.clone(), + inventory_bytes: inventory.clone(), + files: vec![pb::SourceFile { + logical_name: "native/model.yaml".into(), + media_type: "application/yaml".into(), + content: bytes.clone(), + ..Default::default() + }], + })) + .await + .unwrap() + .into_inner() + .snapshot + .unwrap(); + let source = pb::SourceResourceKey { + registration_id: registration.id.clone(), + category: "model".into(), + key: "receipt".into(), + }; + fixture + .service + .create_source_mapping(Request::new(pb::CreateSourceMappingRequest { + registration_revision: registration.revision.clone(), + mapping: Some(pb::SourceMapping { + id: "receipt-title".into(), + source: Some(source.clone()), + selection: Some(pb::SourceSelection { + selection: Some(pb::source_selection::Selection::Entity(reference(receipt))), + }), + ownership: Some(pb::SourceFieldOwnership { + policy: Some(pb::source_field_ownership::Policy::Fields( + pb::SourceOwnedFields { + fields: vec![pb::SourceOwnedField::Title as i32], + }, + )), + }), + adoption: Some(pb::SourceAdoption { + observed: Some(pb::SourceObservedOwner { + owner: Some(reference(receipt)), + uid: uid(receipt), + etag: original.etag.clone(), + }), + authorized_adoption: true, + }), + ..Default::default() + }), + })) + .await + .unwrap(); + let mut payload = receipt.clone(); + let Some(pb::entity::Kind::ReadModel(model)) = &mut payload.kind else { + panic!("read model") + }; + model.title = "Source receipt".into(); + model.doc = "Unowned source documentation".into(); + let preview = fixture + .service + .preview_source_run(Request::new(pb::PreviewSourceRunRequest { + registration_id: registration.id.clone(), + registration_revision: registration.revision.clone(), + snapshot_id: snapshot.id.clone(), + run_id: "receipt-title-update".into(), + native_payloads: vec![pb::SourceNativePayload { + source: Some(source), + entity: Some(payload), + }], + ..Default::default() + })) + .await + .unwrap() + .into_inner() + .run + .unwrap(); + assert_eq!(preview.status, pb::SourceRunStatus::Previewed as i32); + assert_eq!(fixture.get(receipt).await.etag, original.etag); + let request = pb::ApplySourceRunRequest { + registration_id: registration.id.clone(), + run_id: preview.id.clone(), + if_match: preview.revision, + preview_digest: preview.preview_digest, + }; + let applied = fixture + .service + .apply_source_run(Request::new(request.clone())) + .await + .unwrap() + .into_inner() + .run + .unwrap(); + assert_eq!(applied.status, pb::SourceRunStatus::Applied as i32); + let current = fixture.get(receipt).await; + let mut expected = receipt.kind.clone().unwrap(); + let pb::entity::Kind::ReadModel(model) = &mut expected else { + panic!("read model") + }; + model.title = "Source receipt".into(); + let stored = current.entity.as_ref().unwrap(); + assert_eq!(stored.kind.as_ref(), Some(&expected)); + assert_eq!(uid(stored), uid(receipt)); + assert_ne!(current.etag, original.etag); + fixture.remember(stored); + for entity in fixture.expected.values() { + assert_eq!(fixture.get(entity).await.entity.as_ref(), Some(entity)); + } + let replay = fixture + .service + .apply_source_run(Request::new(request)) + .await + .unwrap() + .into_inner() + .run + .unwrap(); + assert_eq!(replay.status, pb::SourceRunStatus::Applied as i32); + assert_eq!(fixture.get(receipt).await.etag, current.etag); + let retained = fixture + .service + .get_source_snapshot(Request::new(pb::GetSourceSnapshotRequest { + registration_id: registration.id, + snapshot_id: snapshot.id, + include_bytes: true, + })) + .await + .unwrap() + .into_inner() + .snapshot + .unwrap(); + assert_eq!(retained.source_version, "00.RC-A"); + assert_eq!(retained.bundle_digest, snapshot.bundle_digest); + assert_eq!(retained.inventory_bytes, inventory); + assert_eq!(retained.files[0].content, bytes); +} + +fn port(variable: &str, default: u16) -> u16 { + std::env::var(variable).map_or(default, |value| { + value.parse().expect("fixture port must be a u16") + }) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "manual same-graph client acceptance; serves loopback gRPC and MCP for at most ten minutes"] +async fn serve_connected_architecture_fixture() { + let fixture = Box::pin(exercise()).await; + let registry = TokenRegistry::from_file_contents( + serde_json::from_value(json!({"principals": { + "fixture-admin": {"role": "admin", "tokens": ["connected-fixture-admin"]} + }})) + .unwrap(), + ) + .unwrap(); + let grpc_listener = tokio::net::TcpListener::bind(( + std::net::Ipv4Addr::LOCALHOST, + port("ATLAS_CONNECTED_FIXTURE_GRPC_PORT", 19473), + )) + .await + .unwrap(); + let mcp_listener = tokio::net::TcpListener::bind(( + std::net::Ipv4Addr::LOCALHOST, + port("ATLAS_CONNECTED_FIXTURE_MCP_PORT", 19474), + )) + .await + .unwrap(); + let grpc_address = grpc_listener.local_addr().unwrap(); + let mcp_address = mcp_listener.local_addr().unwrap(); + let namespace = fixture.namespace.clone(); + let receipt_id = fixture.id("receipt"); + let expected = fixture + .expected + .get(&pb::EntityKey::new(pb::EntityKind::ReadModel, &receipt_id)) + .unwrap() + .clone(); + let service = pb::event_model_service_server::EventModelServiceServer::new(fixture.service) + .max_decoding_message_size(36 * 1024 * 1024) + .max_encoding_message_size(36 * 1024 * 1024); + let grpc = data_plane(service.clone(), BearerAuth::new(registry.clone())); + let (drain, _) = tokio::sync::watch::channel(0i64); + let app = mcp_router( + data_plane(service, BearerAuth::new(registry)), + RpcMetricsLayer::new(ServerMetrics::default()), + InflightLayer::new(Arc::new(AtomicI64::new(0)), Arc::new(drain)), + Some(Duration::from_secs(30)), + 36 * 1024 * 1024, + "127.0.0.1".parse().unwrap(), + ) + .unwrap(); + let seconds = std::env::var("ATLAS_CONNECTED_FIXTURE_SECONDS") + .map_or(600, |value| { + value + .parse::() + .expect("fixture lifetime must be seconds") + }) + .clamp(1, 600); + let mut servers = tokio::task::JoinSet::new(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(seconds); + servers.spawn(async move { + tonic::transport::Server::builder() + .add_service(grpc) + .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new( + grpc_listener, + )) + .await + .unwrap(); + }); + servers.spawn(async move { + axum::serve(mcp_listener, app).await.unwrap(); + }); + let grpc_endpoint = format!("http://{grpc_address}"); + let mut request = Request::new(pb::GetEntityRequest { + kind: pb::EntityKind::ReadModel as i32, + id: Some(receipt_id), + }); + request.metadata_mut().insert( + "authorization", + tonic::metadata::MetadataValue::from_static("Bearer connected-fixture-admin"), + ); + let startup = tokio::time::timeout_at(deadline, async { + let mut client = + pb::event_model_service_client::EventModelServiceClient::connect(grpc_endpoint.clone()) + .await + .map_err(|error| format!("fixture connection failed: {error}"))?; + let stored = client + .get_entity(request) + .await + .map_err(|error| format!("fixture startup read failed: {error}"))? + .into_inner() + .entity; + if stored.as_ref() != Some(&expected) { + return Err("fixture startup read did not match the exercised receipt".to_owned()); + } + Ok(()) + }) + .await; + let outcome = match startup { + Ok(Ok(())) => { + println!( + "READY grpc={grpc_endpoint} mcp=http://{mcp_address}/mcp namespace={namespace} model=model version=1 component=receipt-service token=connected-fixture-admin seconds={seconds}" + ); + tokio::select! { + () = tokio::time::sleep_until(deadline) => Ok(()), + result = servers.join_next() => Err(format!("fixture server exited before deadline: {result:?}")), + result = tokio::signal::ctrl_c() => result.map_err(|error| error.to_string()), + } + } + Ok(Err(error)) => Err(error), + Err(error) => Err(format!("fixture startup deadline elapsed: {error}")), + }; + servers.shutdown().await; + outcome.unwrap_or_else(|error| panic!("{error}")); +} diff --git a/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.test.ts b/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.test.ts index f66aba3..caaa43e 100644 --- a/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.test.ts +++ b/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.test.ts @@ -38,6 +38,8 @@ const ALL_ENTITY_KINDS: EntityKind[] = [ 'person', 'dataProduct', 'specification', + 'decision', + 'document', ]; describe('KIND_SHORT_TO_CAMEL registry completeness', () => { diff --git a/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.ts b/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.ts index 273833a..18becee 100644 --- a/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.ts +++ b/tsworkspace/webapps/studio/src/components/ModelShell.kindmap.ts @@ -35,4 +35,6 @@ export const KIND_SHORT_TO_CAMEL: Record = { person: 'person', data_product: 'dataProduct', specification: 'specification', + decision: 'decision', + document: 'document', }; diff --git a/tsworkspace/webapps/studio/src/components/ModelShell.selection.test.tsx b/tsworkspace/webapps/studio/src/components/ModelShell.selection.test.tsx index a5b19a0..f7fb9db 100644 --- a/tsworkspace/webapps/studio/src/components/ModelShell.selection.test.tsx +++ b/tsworkspace/webapps/studio/src/components/ModelShell.selection.test.tsx @@ -93,6 +93,23 @@ describe('ModelShell selection drawer routing', () => { }); afterEach(cleanup); + it.each([ + ['decision', 'delivery-choice', 'Delivery choice'], + ['document', 'delivery-note', 'Delivery note'], + ] as const)('opens the inspector from a %s focus deep link', async (kind, slug, title) => { + const model = buildModel([{ [kind]: { id: { namespace: 'shop', slug }, title: { value: title } } }]); + const entity = model.entities[0]; + expect(entity.kind).toBe(kind); + rtlRender(, { + wrapper: withNuqsTestingAdapter({ searchParams: `?focus=${kind}:shop/${slug}` }), + }); + + await waitFor(() => expect(screen.getByTestId('inspector').textContent).toBe(title)); + const board = screen.getByTestId('board'); + expect(JSON.parse(board.getAttribute('data-selections') ?? 'null')).toEqual([{ type: 'entity', key: entity.key }]); + expect(JSON.parse(board.getAttribute('data-focus') ?? 'null')).toEqual({ key: entity.key, n: 1 }); + }); + it('passes retained historical context to the inspector before the URL catches up', async () => { const model = buildModel([ { event: { id: { namespace: 'shop', slug: 'order.placed', version: '1' }, title: 'Retained event' } }, From 70239dc4e0aa27d8678c201e3d2dfbb2b9c53477 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sun, 25 Oct 2026 07:37:09 -0400 Subject: [PATCH 2/2] fix(docs): avoid stale fixture endpoint connections Signed-off-by: Yordis Prieto --- docs/how-to/verify-connected-architecture.md | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/docs/how-to/verify-connected-architecture.md b/docs/how-to/verify-connected-architecture.md index dfa5f2e..1589200 100644 --- a/docs/how-to/verify-connected-architecture.md +++ b/docs/how-to/verify-connected-architecture.md @@ -42,20 +42,22 @@ is disposable. Do not seed or restart an existing deployment for this proof. ## Inspect with CLI and MCP -Build the CLI, then replace `` with the READY value: +Build the CLI, then replace `` and `` with the +`namespace=` and `grpc=` values reported in `READY`: ```sh mise exec -- cargo build --manifest-path rsworkspace/Cargo.toml \ -p trogon-atlas-cli --bin trogon-atlas mise exec -- rsworkspace/target/debug/trogon-atlas \ - --endpoint http://127.0.0.1:19473 --auth-token connected-fixture-admin \ + --endpoint '' --auth-token connected-fixture-admin \ --format json describe component//receipt-service@1 mise exec -- rsworkspace/target/debug/trogon-atlas \ - --endpoint http://127.0.0.1:19473 --auth-token connected-fixture-admin \ + --endpoint '' --auth-token connected-fixture-admin \ --format json inspect-data-product data_product//receipts@1 ``` -Connect an MCP client using Streamable HTTP and the same bearer. Initialize +Connect an MCP client to the `mcp=` value reported in `READY`, using +Streamable HTTP and the same bearer. Initialize the connection and discover `tools/list` before invoking tools. Query `get_entity_json` for `automation_slice`, slug `observe`, exact namespace and version `1`. Its equal ReadModel endpoints must retain distinct connection IDs. @@ -83,11 +85,12 @@ new-store selector rebinding or a new source publication through either client. ## Inspect with Studio -Build and start the repository's bridge against the fixture in another terminal: +Build and start the repository's bridge against the fixture in another terminal. +Replace `` with the `grpc=` value reported in `READY`: ```sh mise exec -- pnpm --dir tsworkspace/webapps/studio build -mise exec -- env TROGON_ATLAS_GRPC=http://127.0.0.1:19473 \ +mise exec -- env TROGON_ATLAS_GRPC='' \ TROGON_ATLAS_AUTH_TOKEN=connected-fixture-admin \ TROGON_ATLAS_AUTH_PASSTHROUGH=true PORT=19475 HOST=127.0.0.1 \ node tsworkspace/webapps/studio/server/index.mjs