up

src/Signals/__Tests/StellaOps.Signals.Tests/InMemoryEventsPublisherTests.cs

@@ -3,6 +3,7 @@ using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using StellaOps.Signals.Models;
+using StellaOps.Signals.Options;
 using StellaOps.Signals.Services;
 using Xunit;
 
@@ -14,7 +15,7 @@ public class InMemoryEventsPublisherTests
     public async Task PublishFactUpdatedAsync_EmitsStructuredEvent()
     {
         var logger = new TestLogger<InMemoryEventsPublisher>();
-        var publisher = new InMemoryEventsPublisher(logger);
+        var publisher = new InMemoryEventsPublisher(logger, new SignalsOptions());
 
         var fact = new ReachabilityFactDocument
         {
@@ -23,8 +24,8 @@ public class InMemoryEventsPublisherTests
             ComputedAt = System.DateTimeOffset.Parse("2025-11-18T12:00:00Z"),
             States = new List<ReachabilityStateDocument>
             {
-                new() { Target = "pkg:pypi/django", Reachable = true, Confidence = 0.9 },
-                new() { Target = "pkg:pypi/requests", Reachable = false, Confidence = 0.2 }
+                new() { Target = "pkg:pypi/django", Reachable = true, Confidence = 0.9, Bucket = "runtime", Weight = 0.45 },
+                new() { Target = "pkg:pypi/requests", Reachable = false, Confidence = 0.2, Bucket = "runtime", Weight = 0.45 }
             },
             RuntimeFacts = new List<RuntimeFactDocument>
             {
@@ -40,13 +41,20 @@ public class InMemoryEventsPublisherTests
         Assert.Contains("\"reachableCount\":1", logger.LastMessage);
         Assert.Contains("\"unreachableCount\":1", logger.LastMessage);
         Assert.Contains("\"runtimeFactsCount\":1", logger.LastMessage);
+        Assert.Contains("\"bucket\":\"runtime\"", logger.LastMessage);
+        Assert.Contains("\"weight\":0.45", logger.LastMessage);
+        Assert.Contains("\"factScore\":", logger.LastMessage);
+        Assert.Contains("\"unknownsCount\":0", logger.LastMessage);
+        Assert.Contains("\"unknownsPressure\":0", logger.LastMessage);
+        Assert.Contains("\"stateCount\":2", logger.LastMessage);
+        Assert.Contains("\"targets\":[\"pkg:pypi/django\",\"pkg:pypi/requests\"]", logger.LastMessage);
     }
 
     private sealed class TestLogger<T> : ILogger<T>
     {
         public string LastMessage { get; private set; } = string.Empty;
 
-        public IDisposable BeginScope<TState>(TState state) => NullScope.Instance;
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
 
         public bool IsEnabled(LogLevel logLevel) => true;
 

src/Signals/__Tests/StellaOps.Signals.Tests/ReachabilityScoringServiceTests.cs

@@ -45,6 +45,7 @@ public class ReachabilityScoringServiceTests
 
         var cache = new InMemoryReachabilityCache();
         var eventsPublisher = new RecordingEventsPublisher();
+        var unknowns = new InMemoryUnknownsRepository();
 
         var service = new ReachabilityScoringService(
             callgraphRepository,
@@ -52,6 +53,7 @@ public class ReachabilityScoringServiceTests
             TimeProvider.System,
             Options.Create(options),
             cache,
+            unknowns,
             eventsPublisher,
             NullLogger<ReachabilityScoringService>.Instance);
 
@@ -73,8 +75,13 @@ public class ReachabilityScoringServiceTests
         Assert.Equal("target", state.Target);
         Assert.Equal(new[] { "main", "svc", "target" }, state.Path);
         Assert.Equal(0.9, state.Confidence, 2); // 0.8 + 0.1 runtime bonus
+        Assert.Equal("runtime", state.Bucket);
+        Assert.Equal(0.45, state.Weight, 2);
+        Assert.Equal(0.405, state.Score, 3);
         Assert.Contains("svc", state.Evidence.RuntimeHits);
         Assert.Contains("target", state.Evidence.RuntimeHits);
+
+        Assert.Equal(0.405, fact.Score, 3);
     }
 
     private sealed class InMemoryCallgraphRepository : ICallgraphRepository
@@ -147,4 +154,26 @@ public class ReachabilityScoringServiceTests
             return Task.CompletedTask;
         }
     }
+
+    private sealed class InMemoryUnknownsRepository : IUnknownsRepository
+    {
+        public List<UnknownSymbolDocument> Stored { get; } = new();
+
+        public Task UpsertAsync(string subjectKey, IEnumerable<UnknownSymbolDocument> items, CancellationToken cancellationToken)
+        {
+            Stored.Clear();
+            Stored.AddRange(items);
+            return Task.CompletedTask;
+        }
+
+        public Task<IReadOnlyList<UnknownSymbolDocument>> GetBySubjectAsync(string subjectKey, CancellationToken cancellationToken)
+        {
+            return Task.FromResult((IReadOnlyList<UnknownSymbolDocument>)Stored.ToList());
+        }
+
+        public Task<int> CountBySubjectAsync(string subjectKey, CancellationToken cancellationToken)
+        {
+            return Task.FromResult(Stored.Count);
+        }
+    }
 }

src/Signals/__Tests/StellaOps.Signals.Tests/ReachabilityUnionIngestionServiceTests.cs

@@ -0,0 +1,90 @@
+using System;
+using System.IO;
+using System.IO.Compression;
+using System.Text.Json;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using StellaOps.Signals.Options;
+using StellaOps.Signals.Services;
+using Xunit;
+
+namespace StellaOps.Signals.Tests;
+
+public class ReachabilityUnionIngestionServiceTests
+{
+    [Fact]
+    public async Task IngestAsync_ValidBundle_WritesFilesAndValidatesHashes()
+    {
+        // Arrange
+        var tempRoot = Directory.CreateDirectory(Path.Combine(Path.GetTempPath(), "signals-union-test-" + Guid.NewGuid().ToString("N")));
+        var signalsOptions = new SignalsOptions();
+        signalsOptions.Storage.RootPath = tempRoot.FullName;
+        signalsOptions.Mongo.ConnectionString = "mongodb://localhost";
+        signalsOptions.Mongo.Database = "stub";
+
+        var options = Microsoft.Extensions.Options.Options.Create(signalsOptions);
+
+        using var bundleStream = BuildSampleUnionZip();
+        var service = new ReachabilityUnionIngestionService(NullLogger<ReachabilityUnionIngestionService>.Instance, options);
+
+        // Act
+        var response = await service.IngestAsync("analysis-1", bundleStream, default);
+
+        // Assert
+        Assert.Equal("analysis-1", response.AnalysisId);
+        Assert.Contains(response.Files, f => f.Path == "nodes.ndjson");
+        var metaPath = Path.Combine(tempRoot.FullName, "reachability_graphs", "analysis-1", "meta.json");
+        Assert.True(File.Exists(metaPath));
+
+        // Cleanup
+        tempRoot.Delete(true);
+    }
+
+    private static MemoryStream BuildSampleUnionZip()
+    {
+        var ms = new MemoryStream();
+        using var archive = new ZipArchive(ms, ZipArchiveMode.Create, leaveOpen: true);
+
+        var nodes = archive.CreateEntry("nodes.ndjson");
+        using (var writer = new StreamWriter(nodes.Open()))
+        {
+            writer.WriteLine("{\"symbol_id\":\"sym:dotnet:abc\",\"lang\":\"dotnet\",\"kind\":\"function\",\"display\":\"abc\"}");
+        }
+
+        var edges = archive.CreateEntry("edges.ndjson");
+        using (var writer = new StreamWriter(edges.Open()))
+        {
+            writer.WriteLine("{\"from\":\"sym:dotnet:abc\",\"to\":\"sym:dotnet:def\",\"edge_type\":\"call\",\"source\":{\"origin\":\"static\",\"provenance\":\"il\"}}");
+        }
+
+        // facts_runtime optional left out
+
+        var meta = archive.CreateEntry("meta.json");
+        using (var writer = new StreamWriter(meta.Open()))
+        {
+            var files = new[]
+            {
+                new { path = "nodes.ndjson", sha256 = ComputeSha("{\"symbol_id\":\"sym:dotnet:abc\",\"lang\":\"dotnet\",\"kind\":\"function\",\"display\":\"abc\"}\n"), records = 1 },
+                new { path = "edges.ndjson", sha256 = ComputeSha("{\"from\":\"sym:dotnet:abc\",\"to\":\"sym:dotnet:def\",\"edge_type\":\"call\",\"source\":{\"origin\":\"static\",\"provenance\":\"il\"}}\n"), records = 1 }
+            };
+            var metaObj = new
+            {
+                schema = "reachability-union@0.1",
+                generated_at = "2025-11-23T00:00:00Z",
+                produced_by = new { tool = "test", version = "0.0.1" },
+                files
+            };
+            writer.Write(JsonSerializer.Serialize(metaObj));
+        }
+
+        ms.Position = 0;
+        return ms;
+    }
+
+    private static string ComputeSha(string content)
+    {
+        using var sha = System.Security.Cryptography.SHA256.Create();
+        var bytes = System.Text.Encoding.UTF8.GetBytes(content);
+        return Convert.ToHexString(sha.ComputeHash(bytes)).ToLowerInvariant();
+    }
+}

src/Signals/__Tests/StellaOps.Signals.Tests/UnknownsIngestionServiceTests.cs

@@ -0,0 +1,82 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging.Abstractions;
+using StellaOps.Signals.Models;
+using StellaOps.Signals.Persistence;
+using StellaOps.Signals.Services;
+using Xunit;
+
+namespace StellaOps.Signals.Tests;
+
+public class UnknownsIngestionServiceTests
+{
+    [Fact]
+    public async Task IngestAsync_StoresNormalizedUnknowns()
+    {
+        var repo = new InMemoryUnknownsRepository();
+        var service = new UnknownsIngestionService(repo, TimeProvider.System, NullLogger<UnknownsIngestionService>.Instance);
+
+        var request = new UnknownsIngestRequest
+        {
+            Subject = new ReachabilitySubject { Component = "demo", Version = "1.0.0" },
+            CallgraphId = "cg-1",
+            Unknowns = new List<UnknownSymbolEntry>
+            {
+                new()
+                {
+                    SymbolId = "symA",
+                    Purl = "pkg:pypi/foo",
+                    Reason = "missing-edge"
+                },
+                new() // empty entry should be ignored
+            }
+        };
+
+        var response = await service.IngestAsync(request, CancellationToken.None);
+
+        Assert.Equal("demo|1.0.0", response.SubjectKey);
+        Assert.Equal(1, response.UnknownsCount);
+        Assert.Single(repo.Stored);
+        Assert.Equal("symA", repo.Stored[0].SymbolId);
+        Assert.Equal("pkg:pypi/foo", repo.Stored[0].Purl);
+    }
+
+    [Fact]
+    public async Task IngestAsync_ThrowsWhenEmpty()
+    {
+        var repo = new InMemoryUnknownsRepository();
+        var service = new UnknownsIngestionService(repo, TimeProvider.System, NullLogger<UnknownsIngestionService>.Instance);
+
+        var request = new UnknownsIngestRequest
+        {
+            Subject = new ReachabilitySubject { Component = "demo", Version = "1.0.0" },
+            CallgraphId = "cg-1",
+            Unknowns = new List<UnknownSymbolEntry>()
+        };
+
+        await Assert.ThrowsAsync<UnknownsValidationException>(() => service.IngestAsync(request, CancellationToken.None));
+    }
+
+    private sealed class InMemoryUnknownsRepository : IUnknownsRepository
+    {
+        public List<UnknownSymbolDocument> Stored { get; } = new();
+
+        public Task UpsertAsync(string subjectKey, IEnumerable<UnknownSymbolDocument> items, CancellationToken cancellationToken)
+        {
+            Stored.Clear();
+            Stored.AddRange(items);
+            return Task.CompletedTask;
+        }
+
+        public Task<IReadOnlyList<UnknownSymbolDocument>> GetBySubjectAsync(string subjectKey, CancellationToken cancellationToken)
+        {
+            return Task.FromResult((IReadOnlyList<UnknownSymbolDocument>)Stored);
+        }
+
+        public Task<int> CountBySubjectAsync(string subjectKey, CancellationToken cancellationToken)
+        {
+            return Task.FromResult(Stored.Count);
+        }
+    }
+}