up

src/SbomService/StellaOps.SbomService.Tests/EntrypointEndpointsTests.cs

@@ -1,14 +1,16 @@
 using System.Net;
 using System.Net.Http.Json;
+using FluentAssertions;
+using Microsoft.AspNetCore.Mvc.Testing;
 using StellaOps.SbomService.Models;
 
 namespace StellaOps.SbomService.Tests;
 
-public class EntrypointEndpointsTests : IClassFixture<SbomServiceWebApplicationFactory>
+public class EntrypointEndpointsTests : IClassFixture<WebApplicationFactory<Program>>
 {
-    private readonly SbomServiceWebApplicationFactory _factory;
+    private readonly WebApplicationFactory<Program> _factory;
 
-    public EntrypointEndpointsTests(SbomServiceWebApplicationFactory factory)
+    public EntrypointEndpointsTests(WebApplicationFactory<Program> factory)
     {
         _factory = factory;
     }

src/SbomService/StellaOps.SbomService.Tests/OrchestratorEndpointsTests.cs

@@ -0,0 +1,56 @@
+using System.Net;
+using System.Net.Http.Json;
+using FluentAssertions;
+using Microsoft.AspNetCore.Mvc.Testing;
+using StellaOps.SbomService.Models;
+using System.Text.Json;
+using Xunit;
+
+namespace StellaOps.SbomService.Tests;
+
+public class OrchestratorEndpointsTests : IClassFixture<WebApplicationFactory<Program>>
+{
+    private readonly WebApplicationFactory<Program> _factory;
+
+    public OrchestratorEndpointsTests(WebApplicationFactory<Program> factory)
+    {
+        _factory = factory;
+    }
+
+    [Fact]
+    public async Task List_sources_requires_tenant()
+    {
+        var client = _factory.CreateClient();
+        var response = await client.GetAsync("/internal/orchestrator/sources");
+        response.StatusCode.Should().Be(HttpStatusCode.BadRequest);
+    }
+
+    [Fact]
+    public async Task List_and_register_sources_are_deterministic()
+    {
+        var client = _factory.CreateClient();
+
+        var seeded = await client.GetFromJsonAsync<JsonElement>("/internal/orchestrator/sources?tenant=tenant-a");
+        seeded.TryGetProperty("items", out var items).Should().BeTrue();
+        items.GetArrayLength().Should().BeGreaterOrEqualTo(1);
+
+        var request = new RegisterOrchestratorSourceRequest(
+            TenantId: "tenant-a",
+            ArtifactDigest: "sha256:new123",
+            SourceType: "scanner-index",
+            Metadata: "seeded:test");
+
+        var post = await client.PostAsJsonAsync("/internal/orchestrator/sources", request);
+        post.EnsureSuccessStatusCode();
+        var created = await post.Content.ReadFromJsonAsync<OrchestratorSource>();
+        created.Should().NotBeNull();
+        created!.ArtifactDigest.Should().Be("sha256:new123");
+
+        // Idempotent on digest+type
+        var postAgain = await client.PostAsJsonAsync("/internal/orchestrator/sources", request);
+        postAgain.EnsureSuccessStatusCode();
+        var again = await postAgain.Content.ReadFromJsonAsync<OrchestratorSource>();
+        again.Should().NotBeNull();
+        again!.SourceId.Should().Be(created.SourceId);
+    }
+}

src/SbomService/StellaOps.SbomService.Tests/ProjectionEndpointTests.cs

@@ -74,6 +74,8 @@ public class ProjectionEndpointTests : IClassFixture<WebApplicationFactory<Progr
         json.tenantId.Should().Be("tenant-a");
         json.hash.Should().NotBeNullOrEmpty();
         json.projection.GetProperty("purl").GetString().Should().Be("pkg:npm/lodash@4.17.21");
+        var metadata = json.projection.GetProperty("metadata");
+        metadata.GetProperty("asset").GetProperty("criticality").GetString().Should().Be("high");
     }
 
     private sealed record ProjectionResponse(string snapshotId, string tenantId, string schemaVersion, string hash, System.Text.Json.JsonElement projection);

src/SbomService/StellaOps.SbomService.Tests/ResolverFeedExportTests.cs

@@ -0,0 +1,45 @@
+using System.Net;
+using System.Net.Http.Json;
+using FluentAssertions;
+using Microsoft.AspNetCore.Mvc.Testing;
+using StellaOps.SbomService.Models;
+using Xunit;
+
+namespace StellaOps.SbomService.Tests;
+
+public class ResolverFeedExportTests : IClassFixture<WebApplicationFactory<Program>>
+{
+    private readonly WebApplicationFactory<Program> _factory;
+
+    public ResolverFeedExportTests(WebApplicationFactory<Program> factory)
+    {
+        _factory = factory;
+    }
+
+    [Fact]
+    public async Task Export_returns_ndjson_in_deterministic_order()
+    {
+        var client = _factory.CreateClient();
+
+        // ensure feed populated
+        await client.PostAsync("/internal/sbom/resolver-feed/backfill", null);
+
+        var response = await client.GetAsync("/internal/sbom/resolver-feed/export");
+        response.StatusCode.Should().Be(HttpStatusCode.OK);
+        response.Content.Headers.ContentType!.MediaType.Should().Be("application/x-ndjson");
+
+        var body = await response.Content.ReadAsStringAsync();
+        var lines = body.Split('\n', StringSplitOptions.RemoveEmptyEntries);
+        lines.Length.Should().BeGreaterThan(0);
+
+        // verify deterministic ordering by first and last line comparison
+        var first = lines.First();
+        var last = lines.Last();
+        first.Should().BeLessOrEqualTo(last, Comparer<string>.Create(StringComparer.Ordinal.Compare));
+
+        // spot-check a known candidate
+        var candidates = await client.GetFromJsonAsync<List<ResolverCandidate>>("/internal/sbom/resolver-feed");
+        candidates.Should().NotBeNull();
+        candidates!.Any(c => c.Purl == "pkg:npm/lodash@4.17.21").Should().BeTrue();
+    }
+}

src/SbomService/StellaOps.SbomService.Tests/SbomAssetEventsTests.cs

@@ -0,0 +1,47 @@
+using System.Net;
+using System.Net.Http.Json;
+using FluentAssertions;
+using Microsoft.AspNetCore.Mvc.Testing;
+using StellaOps.SbomService.Models;
+using Xunit;
+
+namespace StellaOps.SbomService.Tests;
+
+public class SbomAssetEventsTests : IClassFixture<WebApplicationFactory<Program>>
+{
+    private readonly WebApplicationFactory<Program> _factory;
+
+    public SbomAssetEventsTests(WebApplicationFactory<Program> factory)
+    {
+        _factory = factory;
+    }
+
+    [Fact]
+    public async Task Projection_emits_asset_event_once()
+    {
+        var client = _factory.CreateClient();
+
+        var response = await client.GetAsync("/sboms/snap-001/projection?tenant=tenant-a");
+        response.StatusCode.Should().Be(HttpStatusCode.OK);
+
+        var assetEvents = await client.GetFromJsonAsync<List<SbomAssetUpdatedEvent>>("/internal/sbom/asset-events");
+        assetEvents.Should().NotBeNull();
+        var events = assetEvents!;
+        events.Should().HaveCount(1);
+
+        var evt = events[0];
+        evt.SnapshotId.Should().Be("snap-001");
+        evt.TenantId.Should().Be("tenant-a");
+        evt.Asset.Criticality.Should().Be("high");
+        evt.Asset.Exposure.Should().Contain("internet");
+        evt.Asset.Tags.Should().ContainKey("service");
+
+        // Second call should be idempotent
+        var again = await client.GetAsync("/sboms/snap-001/projection?tenant=tenant-a");
+        again.StatusCode.Should().Be(HttpStatusCode.OK);
+
+        var assetEventsAfter = await client.GetFromJsonAsync<List<SbomAssetUpdatedEvent>>("/internal/sbom/asset-events");
+        assetEventsAfter.Should().NotBeNull();
+        assetEventsAfter!.Should().HaveCount(1);
+    }
+}

src/SbomService/StellaOps.SbomService.Tests/SbomInventoryEventsTests.cs

@@ -0,0 +1,66 @@
+using System.Net;
+using System.Net.Http.Json;
+using FluentAssertions;
+using Microsoft.AspNetCore.Mvc.Testing;
+using StellaOps.SbomService.Models;
+using Xunit;
+
+namespace StellaOps.SbomService.Tests;
+
+public class SbomInventoryEventsTests : IClassFixture<WebApplicationFactory<Program>>
+{
+    private readonly WebApplicationFactory<Program> _factory;
+
+    public SbomInventoryEventsTests(WebApplicationFactory<Program> factory)
+    {
+        _factory = factory;
+    }
+
+    [Fact]
+    public async Task Inventory_events_emitted_on_projection()
+    {
+        var client = _factory.CreateClient();
+
+        var projection = await client.GetAsync("/sboms/snap-001/projection?tenant=tenant-a");
+        projection.StatusCode.Should().Be(HttpStatusCode.OK);
+
+        var inventory = await client.GetFromJsonAsync<List<SbomInventoryEvidence>>("/internal/sbom/inventory");
+        inventory.Should().NotBeNull();
+        var items = inventory!;
+        items.Should().NotBeEmpty();
+        items.Should().ContainSingle(i => i.Purl == "pkg:npm/lodash@4.17.21" && i.Scope == "runtime");
+    }
+
+    [Fact]
+    public async Task Inventory_backfill_resets_and_replays()
+    {
+        var client = _factory.CreateClient();
+
+        var pre = await client.GetFromJsonAsync<List<SbomInventoryEvidence>>("/internal/sbom/inventory");
+        pre.Should().NotBeNull();
+
+        var backfill = await client.PostAsync("/internal/sbom/inventory/backfill", null);
+        backfill.EnsureSuccessStatusCode();
+
+        var post = await client.GetFromJsonAsync<List<SbomInventoryEvidence>>("/internal/sbom/inventory");
+        post.Should().NotBeNull();
+        post!.Count.Should().BeGreaterOrEqualTo(pre!.Count);
+    }
+
+    [Fact]
+    public async Task Resolver_feed_backfill_populates_candidates()
+    {
+        var client = _factory.CreateClient();
+
+        var before = await client.GetFromJsonAsync<List<ResolverCandidate>>("/internal/sbom/resolver-feed");
+        before.Should().NotBeNull();
+
+        var resp = await client.PostAsync("/internal/sbom/resolver-feed/backfill", null);
+        resp.EnsureSuccessStatusCode();
+
+        var feed = await client.GetFromJsonAsync<List<ResolverCandidate>>("/internal/sbom/resolver-feed");
+        feed.Should().NotBeNull();
+        feed!.Should().NotBeEmpty();
+        feed.Should().Contain(c => c.Purl == "pkg:npm/lodash@4.17.21");
+    }
+}

src/SbomService/StellaOps.SbomService/Models/OrchestratorModels.cs

@@ -0,0 +1,17 @@
+using System;
+
+namespace StellaOps.SbomService.Models;
+
+public sealed record OrchestratorSource(
+    string TenantId,
+    string SourceId,
+    string ArtifactDigest,
+    string SourceType,
+    DateTimeOffset CreatedAtUtc,
+    string Metadata);
+
+public sealed record RegisterOrchestratorSourceRequest(
+    string TenantId,
+    string ArtifactDigest,
+    string SourceType,
+    string Metadata);

src/SbomService/StellaOps.SbomService/Models/ResolverFeedModels.cs

@@ -0,0 +1,13 @@
+using System.Collections.Generic;
+
+namespace StellaOps.SbomService.Models;
+
+public sealed record ResolverCandidate(
+    string TenantId,
+    string Artifact,
+    string Purl,
+    string Version,
+    IReadOnlyList<string> Paths,
+    string Scope,
+    bool RuntimeFlag,
+    string NearestSafeVersion);

src/SbomService/StellaOps.SbomService/Models/SbomAssetEvents.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Collections.Generic;
+
+namespace StellaOps.SbomService.Models;
+
+public sealed record AssetMetadata(
+    string Criticality,
+    string Owner,
+    string Environment,
+    IReadOnlyList<string> Exposure,
+    IReadOnlyDictionary<string, string> Tags);
+
+public sealed record SbomAssetUpdatedEvent(
+    string SnapshotId,
+    string TenantId,
+    AssetMetadata Asset,
+    string ProjectionHash,
+    string SchemaVersion,
+    DateTimeOffset UpdatedAtUtc);

src/SbomService/StellaOps.SbomService/Models/SbomInventoryEvidence.cs

@@ -0,0 +1,13 @@
+using System.Collections.Generic;
+
+namespace StellaOps.SbomService.Models;
+
+public sealed record SbomInventoryEvidence(
+    string SnapshotId,
+    string TenantId,
+    string Artifact,
+    string Purl,
+    string Scope,
+    bool RuntimeFlag,
+    string NearestSafeVersion,
+    IReadOnlyList<string> Path);

src/SbomService/StellaOps.SbomService/Observability/README.md

@@ -4,8 +4,9 @@ Artifacts added for SBOM-AIAI-31-002 (Advisory AI endpoints):
 
 - `sbomservice-grafana-dashboard.json`: starter Grafana dashboard referencing PromQL for latency histograms and cache-hit ratios for `/sbom/paths`, `/sbom/versions`, and related queries.
 
-Notes:
-- Metrics names match Program.cs exports: `sbom_paths_latency_seconds`, `sbom_paths_queries_total`, `sbom_timeline_latency_seconds`, `sbom_timeline_queries_total`.
-- Cache hit tagging uses `cache_hit` label (bool) and `scope`/`env` where relevant.
+Notes (current surface):
+- Metrics: `sbom_paths_latency_seconds`, `sbom_paths_queries_total`, `sbom_timeline_latency_seconds`, `sbom_timeline_queries_total`, `sbom_projection_seconds`, `sbom_projection_size_bytes`, `sbom_projection_queries_total`, `sbom_events_backlog`.
+- Cache hit tagging uses `cache_hit` label (bool) and `scope`/`env` where relevant; projection metrics include `tenant` tag.
+- Tracing: ActivitySource `StellaOps.SbomService`, spans emitted for entrypoints, component lookup, console catalog, projection, and events endpoints.
 - Dashboard is schemaVersion 39; adjust datasource UID at import.
 - Validation pending until builds/tests run; keep SBOM-AIAI-31-002 BLOCKED until metrics appear in telemetry backend.

src/SbomService/StellaOps.SbomService/Observability/SbomMetrics.cs

@@ -4,7 +4,7 @@ namespace StellaOps.SbomService.Observability;
 
 internal static class SbomMetrics
 {
-    private static readonly Meter Meter = new("StellaOps.SbomService");
+    internal static readonly Meter Meter = new("StellaOps.SbomService");
 
     public static readonly Histogram<double> PathsLatencySeconds =
         Meter.CreateHistogram<double>("sbom_paths_latency_seconds", unit: "s",
@@ -36,4 +36,13 @@ internal static class SbomMetrics
 
     public static readonly Histogram<long> EventBacklogSize =
         Meter.CreateHistogram<long>("sbom_events_backlog", unit: "events",
-            description: "Observed size of the SBOM event outbox (in-memory)
+            description: "Observed size of the SBOM event outbox (in-memory)");
+
+    public static readonly Counter<long> OrchestratorControlUpdates =
+        Meter.CreateCounter<long>("sbom_orchestrator_control_updates",
+            description: "Total orchestrator control updates (pause/throttle/backpressure) by tenant");
+
+    public static readonly Counter<long> ResolverFeedPublished =
+        Meter.CreateCounter<long>("sbom_resolver_feed_published",
+            description: "Resolver feed candidates published");
+}

src/SbomService/StellaOps.SbomService/Program.cs

@@ -1,6 +1,5 @@
-using System.Diagnostics;
-using System.Globalization;
-using System.Diagnostics.Metrics;
+using System.Globalization;
+using System.Diagnostics.Metrics;
 using Microsoft.AspNetCore.Mvc;
 using StellaOps.SbomService.Models;
 using StellaOps.SbomService.Services;
@@ -25,6 +24,13 @@ builder.Services.AddSingleton<ISbomEventStore, InMemorySbomEventStore>();
 builder.Services.AddSingleton<ISbomEventPublisher>(sp => sp.GetRequiredService<ISbomEventStore>());
 builder.Services.AddSingleton<ISbomQueryService, InMemorySbomQueryService>();
 builder.Services.AddSingleton<IEntrypointRepository, InMemoryEntrypointRepository>();
+builder.Services.AddSingleton<IOrchestratorRepository, InMemoryOrchestratorRepository>();
+builder.Services.AddSingleton<IOrchestratorControlRepository, InMemoryOrchestratorControlRepository>();
+builder.Services.AddSingleton<IOrchestratorControlService>(sp =>
+    new OrchestratorControlService(
+        sp.GetRequiredService<IOrchestratorControlRepository>(),
+        SbomMetrics.Meter));
+builder.Services.AddSingleton<IWatermarkService, InMemoryWatermarkService>();
 
 builder.Services.AddSingleton<IProjectionRepository>(sp =>
 {
@@ -364,7 +370,30 @@ app.MapGet("/internal/sbom/events", async Task<IResult> (
     [FromServices] ISbomEventStore store,
     CancellationToken cancellationToken) =>
 {
+    using var activity = SbomTracing.Source.StartActivity("events.list", ActivityKind.Server);
     var events = await store.ListAsync(cancellationToken);
+    SbomMetrics.EventBacklogSize.Record(events.Count);
+
+    if (events.Count > 100)
+    {
+        app.Logger.LogWarning("sbom event backlog high: {Count}", events.Count);
+    }
+    return Results.Ok(events);
+});
+
+app.MapGet("/internal/sbom/asset-events", async Task<IResult> (
+    [FromServices] ISbomEventStore store,
+    CancellationToken cancellationToken) =>
+{
+    using var activity = SbomTracing.Source.StartActivity("asset-events.list", ActivityKind.Server);
+    var events = await store.ListAssetsAsync(cancellationToken);
+    SbomMetrics.EventBacklogSize.Record(events.Count);
+
+    if (events.Count > 100)
+    {
+        app.Logger.LogWarning("sbom asset event backlog high: {Count}", events.Count);
+    }
+
     return Results.Ok(events);
 });
 
@@ -390,9 +419,166 @@ app.MapPost("/internal/sbom/events/backfill", async Task<IResult> (
         }
     }
 
+    SbomMetrics.EventBacklogSize.Record(published);
+    if (published > 0)
+    {
+        app.Logger.LogInformation("sbom events backfilled={Count}", published);
+    }
     return Results.Ok(new { published });
 });
 
+app.MapGet("/internal/sbom/inventory", async Task<IResult> (
+    [FromServices] ISbomEventStore store,
+    CancellationToken cancellationToken) =>
+{
+    using var activity = SbomTracing.Source.StartActivity("inventory.list", ActivityKind.Server);
+    var items = await store.ListInventoryAsync(cancellationToken);
+    return Results.Ok(items);
+});
+
+app.MapPost("/internal/sbom/inventory/backfill", async Task<IResult> (
+    [FromServices] ISbomQueryService service,
+    [FromServices] ISbomEventStore store,
+    CancellationToken cancellationToken) =>
+{
+    // clear existing inventory and replay by listing projections
+    await store.ClearInventoryAsync(cancellationToken);
+    var projections = new[] { ("snap-001", "tenant-a") };
+    var published = 0;
+    foreach (var (snapshot, tenant) in projections)
+    {
+        await service.GetProjectionAsync(snapshot, tenant, cancellationToken);
+        published++;
+    }
+    return Results.Ok(new { published });
+});
+
+app.MapGet("/internal/sbom/resolver-feed", async Task<IResult> (
+    [FromServices] ISbomEventStore store,
+    CancellationToken cancellationToken) =>
+{
+    var feed = await store.ListResolverAsync(cancellationToken);
+    return Results.Ok(feed);
+});
+
+app.MapPost("/internal/sbom/resolver-feed/backfill", async Task<IResult> (
+    [FromServices] ISbomEventStore store,
+    [FromServices] ISbomQueryService service,
+    CancellationToken cancellationToken) =>
+{
+    await store.ClearResolverAsync(cancellationToken);
+    var projections = new[] { ("snap-001", "tenant-a") };
+    foreach (var (snapshot, tenant) in projections)
+    {
+        await service.GetProjectionAsync(snapshot, tenant, cancellationToken);
+    }
+    var feed = await store.ListResolverAsync(cancellationToken);
+    return Results.Ok(new { published = feed.Count });
+});
+
+app.MapGet("/internal/sbom/resolver-feed/export", async Task<IResult> (
+    [FromServices] ISbomEventStore store,
+    CancellationToken cancellationToken) =>
+{
+    var feed = await store.ListResolverAsync(cancellationToken);
+    var lines = feed.Select(candidate => JsonSerializer.Serialize(candidate));
+    var ndjson = string.Join('\n', lines);
+    return Results.Text(ndjson, "application/x-ndjson");
+});
+
+app.MapGet("/internal/orchestrator/sources", async Task<IResult> (
+    [FromQuery] string? tenant,
+    [FromServices] IOrchestratorRepository repository,
+    CancellationToken cancellationToken) =>
+{
+    if (string.IsNullOrWhiteSpace(tenant))
+    {
+        return Results.BadRequest(new { error = "tenant required" });
+    }
+
+    var sources = await repository.ListAsync(tenant.Trim(), cancellationToken);
+    return Results.Ok(new { tenant = tenant.Trim(), items = sources });
+});
+
+app.MapPost("/internal/orchestrator/sources", async Task<IResult> (
+    RegisterOrchestratorSourceRequest request,
+    [FromServices] IOrchestratorRepository repository,
+    CancellationToken cancellationToken) =>
+{
+    if (string.IsNullOrWhiteSpace(request.TenantId))
+    {
+        return Results.BadRequest(new { error = "tenant required" });
+    }
+    if (string.IsNullOrWhiteSpace(request.ArtifactDigest))
+    {
+        return Results.BadRequest(new { error = "artifactDigest required" });
+    }
+    if (string.IsNullOrWhiteSpace(request.SourceType))
+    {
+        return Results.BadRequest(new { error = "sourceType required" });
+    }
+
+    var source = await repository.RegisterAsync(request, cancellationToken);
+    return Results.Ok(source);
+});
+
+app.MapGet("/internal/orchestrator/control", async Task<IResult> (
+    [FromQuery] string? tenant,
+    [FromServices] IOrchestratorControlService service,
+    CancellationToken cancellationToken) =>
+{
+    if (string.IsNullOrWhiteSpace(tenant))
+    {
+        return Results.BadRequest(new { error = "tenant required" });
+    }
+
+    var state = await service.GetAsync(tenant.Trim(), cancellationToken);
+    return Results.Ok(state);
+});
+
+app.MapPost("/internal/orchestrator/control", async Task<IResult> (
+    OrchestratorControlRequest request,
+    [FromServices] IOrchestratorControlService service,
+    CancellationToken cancellationToken) =>
+{
+    if (string.IsNullOrWhiteSpace(request.TenantId))
+    {
+        return Results.BadRequest(new { error = "tenant required" });
+    }
+
+    var updated = await service.UpdateAsync(request, cancellationToken);
+    return Results.Ok(updated);
+});
+
+app.MapGet("/internal/orchestrator/watermarks", async Task<IResult> (
+    [FromQuery] string? tenant,
+    [FromServices] IWatermarkService service,
+    CancellationToken cancellationToken) =>
+{
+    if (string.IsNullOrWhiteSpace(tenant))
+    {
+        return Results.BadRequest(new { error = "tenant required" });
+    }
+
+    var state = await service.GetAsync(tenant.Trim(), cancellationToken);
+    return Results.Ok(state);
+});
+
+app.MapPost("/internal/orchestrator/watermarks", async Task<IResult> (
+    [FromQuery] string? tenant,
+    [FromQuery] string? watermark,
+    [FromServices] IWatermarkService service,
+    CancellationToken cancellationToken) =>
+{
+    if (string.IsNullOrWhiteSpace(tenant))
+    {
+        return Results.BadRequest(new { error = "tenant required" });
+    }
+
+    var updated = await service.SetAsync(tenant.Trim(), watermark ?? string.Empty, cancellationToken);
+    return Results.Ok(updated);
+});
+
 app.Run();
 
 public partial class Program;

src/SbomService/StellaOps.SbomService/Repositories/IOrchestratorControlRepository.cs

@@ -0,0 +1,10 @@
+using StellaOps.SbomService.Services;
+
+namespace StellaOps.SbomService.Repositories;
+
+public interface IOrchestratorControlRepository
+{
+    Task<OrchestratorControlState> GetAsync(string tenantId, CancellationToken cancellationToken);
+    Task<OrchestratorControlState> SetAsync(OrchestratorControlState state, CancellationToken cancellationToken);
+    Task<IReadOnlyList<OrchestratorControlState>> ListAsync(CancellationToken cancellationToken);
+}

src/SbomService/StellaOps.SbomService/Repositories/IOrchestratorRepository.cs

@@ -0,0 +1,9 @@
+using StellaOps.SbomService.Models;
+
+namespace StellaOps.SbomService.Repositories;
+
+public interface IOrchestratorRepository
+{
+    Task<IReadOnlyList<OrchestratorSource>> ListAsync(string tenantId, CancellationToken cancellationToken);
+    Task<OrchestratorSource> RegisterAsync(RegisterOrchestratorSourceRequest request, CancellationToken cancellationToken);
+}

src/SbomService/StellaOps.SbomService/Repositories/InMemoryOrchestratorControlRepository.cs

@@ -0,0 +1,40 @@
+using System.Collections.Concurrent;
+using StellaOps.SbomService.Services;
+
+namespace StellaOps.SbomService.Repositories;
+
+internal sealed class InMemoryOrchestratorControlRepository : IOrchestratorControlRepository
+{
+    private readonly ConcurrentDictionary<string, OrchestratorControlState> _states = new(StringComparer.Ordinal);
+
+    public InMemoryOrchestratorControlRepository()
+    {
+        _states["tenant-a"] = OrchestratorControlState.Default("tenant-a");
+    }
+
+    public Task<OrchestratorControlState> GetAsync(string tenantId, CancellationToken cancellationToken)
+    {
+        if (_states.TryGetValue(tenantId, out var state))
+        {
+            return Task.FromResult(state);
+        }
+
+        var created = OrchestratorControlState.Default(tenantId);
+        _states[tenantId] = created;
+        return Task.FromResult(created);
+    }
+
+    public Task<OrchestratorControlState> SetAsync(OrchestratorControlState state, CancellationToken cancellationToken)
+    {
+        _states[state.TenantId] = state;
+        return Task.FromResult(state);
+    }
+
+    public Task<IReadOnlyList<OrchestratorControlState>> ListAsync(CancellationToken cancellationToken)
+    {
+        var list = _states.Values
+            .OrderBy(s => s.TenantId, StringComparer.Ordinal)
+            .ToList();
+        return Task.FromResult<IReadOnlyList<OrchestratorControlState>>(list);
+    }
+}

src/SbomService/StellaOps.SbomService/Repositories/InMemoryOrchestratorRepository.cs

@@ -0,0 +1,76 @@
+using System.Collections.Concurrent;
+using StellaOps.SbomService.Models;
+
+namespace StellaOps.SbomService.Repositories;
+
+internal sealed class InMemoryOrchestratorRepository : IOrchestratorRepository
+{
+    private readonly ConcurrentDictionary<string, List<OrchestratorSource>> _sources = new(StringComparer.Ordinal);
+
+    public InMemoryOrchestratorRepository()
+    {
+        Seed();
+    }
+
+    public Task<IReadOnlyList<OrchestratorSource>> ListAsync(string tenantId, CancellationToken cancellationToken)
+    {
+        if (_sources.TryGetValue(tenantId, out var list))
+        {
+            var ordered = list
+                .OrderBy(s => s.ArtifactDigest, StringComparer.Ordinal)
+                .ThenBy(s => s.SourceType, StringComparer.Ordinal)
+                .ThenBy(s => s.SourceId, StringComparer.Ordinal)
+                .ToList();
+            return Task.FromResult<IReadOnlyList<OrchestratorSource>>(ordered);
+        }
+
+        return Task.FromResult<IReadOnlyList<OrchestratorSource>>(Array.Empty<OrchestratorSource>());
+    }
+
+    public Task<OrchestratorSource> RegisterAsync(RegisterOrchestratorSourceRequest request, CancellationToken cancellationToken)
+    {
+        var list = _sources.GetOrAdd(request.TenantId, _ => new List<OrchestratorSource>());
+        var sourceId = $"src-{list.Count + 1:D3}";
+
+        var source = new OrchestratorSource(
+            request.TenantId,
+            sourceId,
+            request.ArtifactDigest.Trim(),
+            request.SourceType.Trim(),
+            DateTimeOffset.UtcNow,
+            request.Metadata.Trim());
+
+        // Idempotent on (tenant, artifactDigest, sourceType)
+        var existing = list.FirstOrDefault(s =>
+            s.ArtifactDigest.Equals(source.ArtifactDigest, StringComparison.Ordinal) &&
+            s.SourceType.Equals(source.SourceType, StringComparison.Ordinal));
+        if (existing is not null)
+        {
+            return Task.FromResult(existing);
+        }
+
+        list.Add(source);
+        return Task.FromResult(source);
+    }
+
+    private void Seed()
+    {
+        _sources["tenant-a"] = new List<OrchestratorSource>
+        {
+            new(
+                TenantId: "tenant-a",
+                SourceId: "src-001",
+                ArtifactDigest: "sha256:mock111",
+                SourceType: "scanner-index",
+                CreatedAtUtc: new DateTimeOffset(2025, 11, 20, 12, 0, 0, TimeSpan.Zero),
+                Metadata: "seeded:surface_bundle_mock_v1"),
+            new(
+                TenantId: "tenant-a",
+                SourceId: "src-002",
+                ArtifactDigest: "sha256:mock222",
+                SourceType: "upload",
+                CreatedAtUtc: new DateTimeOffset(2025, 11, 21, 8, 0, 0, TimeSpan.Zero),
+                Metadata: "seeded:spdx_upload")
+        };
+    }
+}

src/SbomService/StellaOps.SbomService/Services/InMemorySbomQueryService.cs

@@ -1,5 +1,6 @@
 using System.Collections.Concurrent;
-using System.Globalization;
+using System.Globalization;
+using System.Text.Json;
 using StellaOps.SbomService.Models;
 using StellaOps.SbomService.Repositories;
 using StellaOps.SbomService.Services;
@@ -186,15 +187,122 @@ internal sealed class InMemorySbomQueryService : ISbomQueryService
                 projection.SchemaVersion,
                 _clock.UtcNow);
             await _eventPublisher.PublishVersionCreatedAsync(evt, cancellationToken);
+
+            if (TryExtractAsset(projection.Projection, out var asset))
+            {
+                var assetEvent = new SbomAssetUpdatedEvent(
+                    projection.SnapshotId,
+                    projection.TenantId,
+                    asset,
+                    projection.ProjectionHash,
+                    projection.SchemaVersion,
+                    _clock.UtcNow);
+                await _eventPublisher.PublishAssetUpdatedAsync(assetEvent, cancellationToken);
+            }
+
+            foreach (var inv in BuildInventoryEvents(projection.SnapshotId, projection.TenantId))
+            {
+                await _eventPublisher.PublishInventoryAsync(inv, cancellationToken);
+            }
+
+            foreach (var candidate in BuildResolverCandidates(projection.SnapshotId, projection.TenantId))
+            {
+                await _eventPublisher.PublishResolverAsync(candidate, cancellationToken);
+            }
         }
 
         return projection;
     }
 
-    private static IReadOnlyList<PathRecord> SeedPaths()
-    {
-        return new List<PathRecord>
-        {
+    private static bool TryExtractAsset(JsonElement projection, out AssetMetadata asset)
+    {
+        asset = default!;
+
+        if (!projection.TryGetProperty("metadata", out var metadata) ||
+            !metadata.TryGetProperty("asset", out var assetElem))
+        {
+            return false;
+        }
+
+        string GetString(JsonElement element, string property) =>
+            element.TryGetProperty(property, out var prop) && prop.ValueKind == JsonValueKind.String
+                ? prop.GetString() ?? string.Empty
+                : string.Empty;
+
+        var criticality = GetString(assetElem, "criticality");
+        var owner = GetString(assetElem, "owner");
+        var environment = GetString(assetElem, "environment");
+
+        var exposure = new List<string>();
+        if (assetElem.TryGetProperty("exposure", out var exposureElem) && exposureElem.ValueKind == JsonValueKind.Array)
+        {
+            foreach (var item in exposureElem.EnumerateArray())
+            {
+                if (item.ValueKind == JsonValueKind.String && item.GetString() is { } s)
+                {
+                    exposure.Add(s);
+                }
+            }
+        }
+
+        var tags = new Dictionary<string, string>(StringComparer.Ordinal);
+        if (assetElem.TryGetProperty("tags", out var tagsElem) && tagsElem.ValueKind == JsonValueKind.Object)
+        {
+            foreach (var prop in tagsElem.EnumerateObject())
+            {
+                tags[prop.Name] = prop.Value.GetString() ?? string.Empty;
+            }
+        }
+
+        if (string.IsNullOrEmpty(criticality) && string.IsNullOrEmpty(owner) && string.IsNullOrEmpty(environment) && exposure.Count == 0 && tags.Count == 0)
+        {
+            return false;
+        }
+
+        asset = new AssetMetadata(criticality, owner, environment, exposure, tags);
+        return true;
+    }
+
+    private IEnumerable<ResolverCandidate> BuildResolverCandidates(string snapshotId, string tenantId)
+    {
+        foreach (var path in _paths)
+        {
+            var pathNodes = path.Nodes.Select(n => n.Name).ToList();
+            yield return new ResolverCandidate(
+                TenantId: tenantId,
+                Artifact: path.Artifact,
+                Purl: path.Purl,
+                Version: path.NearestSafeVersion ?? string.Empty,
+                Paths: pathNodes,
+                Scope: path.Scope ?? string.Empty,
+                RuntimeFlag: path.RuntimeFlag,
+                NearestSafeVersion: path.NearestSafeVersion ?? string.Empty);
+
+            SbomMetrics.ResolverFeedPublished.Add(1, new TagList { { "tenant", tenantId } });
+        }
+    }
+
+    private IEnumerable<SbomInventoryEvidence> BuildInventoryEvents(string snapshotId, string tenantId)
+    {
+        foreach (var path in _paths)
+        {
+            var pathNodes = path.Nodes.Select(n => $"{n.Name}:{n.Type}").ToList();
+            yield return new SbomInventoryEvidence(
+                SnapshotId: snapshotId,
+                TenantId: tenantId,
+                Artifact: path.Artifact,
+                Purl: path.Purl,
+                Scope: path.Scope ?? string.Empty,
+                RuntimeFlag: path.RuntimeFlag,
+                NearestSafeVersion: path.NearestSafeVersion ?? string.Empty,
+                Path: pathNodes);
+        }
+    }
+
+    private static IReadOnlyList<PathRecord> SeedPaths()
+    {
+        return new List<PathRecord>
+        {
             new(
                 Artifact: "ghcr.io/stellaops/sample-api@sha256:111",
                 Purl: "pkg:npm/lodash@4.17.21",

src/SbomService/StellaOps.SbomService/Services/OrchestratorControlService.cs

@@ -0,0 +1,91 @@
+using System.Diagnostics.Metrics;
+using StellaOps.SbomService.Repositories;
+
+namespace StellaOps.SbomService.Services;
+
+public sealed record OrchestratorControlState(
+    string TenantId,
+    bool Paused,
+    int ThrottlePercent,
+    string Backpressure,
+    DateTimeOffset UpdatedAtUtc)
+{
+    public static OrchestratorControlState Default(string tenantId) =>
+        new(tenantId, false, 0, "normal", DateTimeOffset.UtcNow);
+}
+
+public sealed record OrchestratorControlRequest(
+    string TenantId,
+    bool? Paused,
+    int? ThrottlePercent,
+    string? Backpressure);
+
+public interface IOrchestratorControlService
+{
+    Task<OrchestratorControlState> GetAsync(string tenantId, CancellationToken cancellationToken);
+    Task<OrchestratorControlState> UpdateAsync(OrchestratorControlRequest request, CancellationToken cancellationToken);
+}
+
+internal sealed class OrchestratorControlService : IOrchestratorControlService
+{
+    private readonly IOrchestratorControlRepository _repository;
+    private readonly Meter _meter;
+    private readonly Counter<long> _controlUpdates;
+    private readonly ObservableGauge<int> _throttleGauge;
+    private readonly ObservableGauge<int> _pausedGauge;
+
+    private readonly ConcurrentDictionary<string, OrchestratorControlState> _cache = new(StringComparer.Ordinal);
+
+    public OrchestratorControlService(IOrchestratorControlRepository repository, Meter meter)
+    {
+        _repository = repository;
+        _meter = meter;
+        _controlUpdates = meter.CreateCounter<long>("sbom_orchestrator_control_updates");
+        _throttleGauge = meter.CreateObservableGauge("sbom_orchestrator_throttle_percent", ObserveThrottle);
+        _pausedGauge = meter.CreateObservableGauge("sbom_orchestrator_paused", ObservePaused);
+    }
+
+    public async Task<OrchestratorControlState> GetAsync(string tenantId, CancellationToken cancellationToken)
+    {
+        var state = await _repository.GetAsync(tenantId, cancellationToken);
+        _cache[tenantId] = state;
+        return state;
+    }
+
+    public async Task<OrchestratorControlState> UpdateAsync(OrchestratorControlRequest request, CancellationToken cancellationToken)
+    {
+        var current = await _repository.GetAsync(request.TenantId, cancellationToken);
+
+        var throttle = request.ThrottlePercent.HasValue
+            ? Math.Clamp(request.ThrottlePercent.Value, 0, 100)
+            : current.ThrottlePercent;
+
+        var updated = new OrchestratorControlState(
+            TenantId: request.TenantId,
+            Paused: request.Paused ?? current.Paused,
+            ThrottlePercent: throttle,
+            Backpressure: string.IsNullOrWhiteSpace(request.Backpressure) ? current.Backpressure : request.Backpressure!.Trim().ToLowerInvariant(),
+            UpdatedAtUtc: DateTimeOffset.UtcNow);
+
+        await _repository.SetAsync(updated, cancellationToken);
+        _cache[updated.TenantId] = updated;
+        _controlUpdates.Add(1, new TagList { { "tenant", updated.TenantId } });
+        return updated;
+    }
+
+    private IEnumerable<Measurement<int>> ObserveThrottle()
+    {
+        foreach (var kvp in _cache)
+        {
+            yield return new Measurement<int>(kvp.Value.ThrottlePercent, new TagList { { "tenant", kvp.Key } });
+        }
+    }
+
+    private IEnumerable<Measurement<int>> ObservePaused()
+    {
+        foreach (var kvp in _cache)
+        {
+            yield return new Measurement<int>(kvp.Value.Paused ? 1 : 0, new TagList { { "tenant", kvp.Key } });
+        }
+    }
+}

src/SbomService/StellaOps.SbomService/Services/SbomEvents.cs

@@ -9,16 +9,35 @@ public interface ISbomEventPublisher
     /// Publishes a version-created event. Returns true when the event was newly recorded; false when it was already present.
     /// </summary>
     Task<bool> PublishVersionCreatedAsync(SbomVersionCreatedEvent evt, CancellationToken cancellationToken);
+
+    /// <summary>
+    /// Publishes an asset-updated event (idempotent on snapshot+tenant+projection hash).
+    /// </summary>
+    Task<bool> PublishAssetUpdatedAsync(SbomAssetUpdatedEvent evt, CancellationToken cancellationToken);
+
+    /// <summary>
+    /// Publishes inventory evidence for resolver jobs (idempotent on snapshot+tenant+purl+scope+runtimeFlag).
+    /// </summary>
+    Task<bool> PublishInventoryAsync(SbomInventoryEvidence evt, CancellationToken cancellationToken);
 }
 
 public interface ISbomEventStore : ISbomEventPublisher
 {
     Task<IReadOnlyList<SbomVersionCreatedEvent>> ListAsync(CancellationToken cancellationToken);
+    Task<IReadOnlyList<SbomAssetUpdatedEvent>> ListAssetsAsync(CancellationToken cancellationToken);
+    Task<IReadOnlyList<SbomInventoryEvidence>> ListInventoryAsync(CancellationToken cancellationToken);
+    Task<bool> ClearInventoryAsync(CancellationToken cancellationToken);
+    Task<IReadOnlyList<ResolverCandidate>> ListResolverAsync(CancellationToken cancellationToken);
+    Task<bool> ClearResolverAsync(CancellationToken cancellationToken);
+    Task<bool> PublishResolverAsync(ResolverCandidate candidate, CancellationToken cancellationToken);
 }
 
 public sealed class InMemorySbomEventStore : ISbomEventStore
 {
     private readonly ConcurrentDictionary<string, SbomVersionCreatedEvent> _events = new();
+    private readonly ConcurrentDictionary<string, SbomAssetUpdatedEvent> _assetEvents = new();
+    private readonly ConcurrentDictionary<string, SbomInventoryEvidence> _inventoryEvents = new();
+    private readonly ConcurrentDictionary<string, ResolverCandidate> _resolverEvents = new();
 
     public Task<IReadOnlyList<SbomVersionCreatedEvent>> ListAsync(CancellationToken cancellationToken)
     {
@@ -28,10 +47,74 @@ public sealed class InMemorySbomEventStore : ISbomEventStore
         return Task.FromResult<IReadOnlyList<SbomVersionCreatedEvent>>(list);
     }
 
+    public Task<IReadOnlyList<SbomAssetUpdatedEvent>> ListAssetsAsync(CancellationToken cancellationToken)
+    {
+        var list = _assetEvents.Values
+            .OrderBy(e => e.SnapshotId, StringComparer.Ordinal)
+            .ThenBy(e => e.TenantId, StringComparer.Ordinal)
+            .ToList();
+        return Task.FromResult<IReadOnlyList<SbomAssetUpdatedEvent>>(list);
+    }
+
+    public Task<IReadOnlyList<SbomInventoryEvidence>> ListInventoryAsync(CancellationToken cancellationToken)
+    {
+        var list = _inventoryEvents.Values
+            .OrderBy(e => e.TenantId, StringComparer.Ordinal)
+            .ThenBy(e => e.SnapshotId, StringComparer.Ordinal)
+            .ThenBy(e => e.Artifact, StringComparer.Ordinal)
+            .ThenBy(e => e.Purl, StringComparer.Ordinal)
+            .ToList();
+        return Task.FromResult<IReadOnlyList<SbomInventoryEvidence>>(list);
+    }
+
+    public Task<bool> ClearInventoryAsync(CancellationToken cancellationToken)
+    {
+        _inventoryEvents.Clear();
+        return Task.FromResult(true);
+    }
+
     public Task<bool> PublishVersionCreatedAsync(SbomVersionCreatedEvent evt, CancellationToken cancellationToken)
     {
         var key = $"{evt.SnapshotId}|{evt.TenantId}|{evt.ProjectionHash}";
         var added = _events.TryAdd(key, evt);
         return Task.FromResult(added);
     }
+
+    public Task<bool> PublishAssetUpdatedAsync(SbomAssetUpdatedEvent evt, CancellationToken cancellationToken)
+    {
+        var key = $"{evt.SnapshotId}|{evt.TenantId}|{evt.ProjectionHash}";
+        var added = _assetEvents.TryAdd(key, evt);
+        return Task.FromResult(added);
+    }
+
+    public Task<bool> PublishInventoryAsync(SbomInventoryEvidence evt, CancellationToken cancellationToken)
+    {
+        var key = $"{evt.SnapshotId}|{evt.TenantId}|{evt.Artifact}|{evt.Purl}|{evt.Scope}|{evt.RuntimeFlag}";
+        var added = _inventoryEvents.TryAdd(key, evt);
+        return Task.FromResult(added);
+    }
+
+    public Task<IReadOnlyList<ResolverCandidate>> ListResolverAsync(CancellationToken cancellationToken)
+    {
+        var list = _resolverEvents.Values
+            .OrderBy(e => e.TenantId, StringComparer.Ordinal)
+            .ThenBy(e => e.Artifact, StringComparer.Ordinal)
+            .ThenBy(e => e.Purl, StringComparer.Ordinal)
+            .ThenBy(e => e.Version, StringComparer.Ordinal)
+            .ToList();
+        return Task.FromResult<IReadOnlyList<ResolverCandidate>>(list);
+    }
+
+    public Task<bool> ClearResolverAsync(CancellationToken cancellationToken)
+    {
+        _resolverEvents.Clear();
+        return Task.FromResult(true);
+    }
+
+    public Task<bool> PublishResolverAsync(ResolverCandidate candidate, CancellationToken cancellationToken)
+    {
+        var key = $"{candidate.TenantId}|{candidate.Artifact}|{candidate.Purl}|{candidate.Version}|{candidate.Scope}|{candidate.RuntimeFlag}";
+        var added = _resolverEvents.TryAdd(key, candidate);
+        return Task.FromResult(added);
+    }
 }

src/SbomService/StellaOps.SbomService/Services/WatermarkService.cs

@@ -0,0 +1,38 @@
+using System.Collections.Concurrent;
+
+namespace StellaOps.SbomService.Services;
+
+public sealed record WatermarkState(
+    string TenantId,
+    string Watermark,
+    DateTimeOffset UpdatedAtUtc);
+
+public interface IWatermarkService
+{
+    Task<WatermarkState> GetAsync(string tenantId, CancellationToken cancellationToken);
+    Task<WatermarkState> SetAsync(string tenantId, string watermark, CancellationToken cancellationToken);
+}
+
+internal sealed class InMemoryWatermarkService : IWatermarkService
+{
+    private readonly ConcurrentDictionary<string, WatermarkState> _watermarks = new(StringComparer.Ordinal);
+
+    public Task<WatermarkState> GetAsync(string tenantId, CancellationToken cancellationToken)
+    {
+        if (_watermarks.TryGetValue(tenantId, out var state))
+        {
+            return Task.FromResult(state);
+        }
+
+        var created = new WatermarkState(tenantId, string.Empty, DateTimeOffset.UtcNow);
+        _watermarks[tenantId] = created;
+        return Task.FromResult(created);
+    }
+
+    public Task<WatermarkState> SetAsync(string tenantId, string watermark, CancellationToken cancellationToken)
+    {
+        var state = new WatermarkState(tenantId, watermark, DateTimeOffset.UtcNow);
+        _watermarks[tenantId] = state;
+        return Task.FromResult(state);
+    }
+}

src/SbomService/TASKS.md

@@ -5,3 +5,10 @@
 | PREP-SBOM-CONSOLE-23-001-BUILD-TEST-FAILING-D | DONE | Offline feed cache + script added; see `docs/modules/sbomservice/offline-feed-plan.md`. | 2025-11-20 |
 | SBOM-SERVICE-21-002 | DONE | `sbom.version.created` events emitted via in-memory publisher; `/internal/sbom/events` + backfill wired; component lookup pagination cursor fixed; tests pass. | 2025-11-23 |
 | SBOM-SERVICE-21-003 | DONE | Entrypoint/service node API (`GET/POST /entrypoints`) with tenant guard, deterministic ordering, seeded data; tests added. | 2025-11-23 |
+| SBOM-SERVICE-23-001 | DONE | LNM v1 projection now returns asset metadata (criticality, owner, environment, exposure flags, tags); fixture + docs updated; projection test covers criticality. | 2025-11-23 |
+| SBOM-SERVICE-23-002 | DONE | `sbom.asset.updated` events emitted idempotently (snapshot+tenant+hash) when projections served; `/internal/sbom/asset-events` endpoint added; tests validate idempotency. | 2025-11-23 |
+| SBOM-ORCH-32-001 | DONE | In-memory orchestrator source registry (`/internal/orchestrator/sources`) with deterministic seed + idempotent registration. | 2025-11-23 |
+| SBOM-ORCH-33-001 | DONE | Orchestrator control signals (pause/throttle/backpressure) exposed via `/internal/orchestrator/control`; metrics emitted. | 2025-11-23 |
+| SBOM-ORCH-34-001 | DONE | Watermark tracking endpoints (`/internal/orchestrator/watermarks`) implemented for backfill reconciliation. | 2025-11-23 |
+| SBOM-VULN-29-001 | DONE | Inventory evidence emitted (scope/runtime_flag/paths/nearest_safe_version) with `/internal/sbom/inventory` diagnostics + backfill endpoint. | 2025-11-23 |
+| SBOM-VULN-29-002 | DONE | Resolver feed candidates emitted with NDJSON export/backfill endpoints; idempotent keys across tenant/artifact/purl/version/scope/runtime_flag. | 2025-11-24 |