From d0c21e8ee8f852c90f2995880433f5090b99e90d Mon Sep 17 00:00:00 2001 From: ancplua Date: Sun, 9 Aug 2026 02:02:37 +0200 Subject: [PATCH 1/2] Add agent diagnostic snapshot capture --- eng/config/collector-semantic-policy.json | 16 +- .../Qyl.Cli/Codex/ActiveWorkflowRunStore.cs | 3 + .../Qyl.Cli/Codex/CodexEventNormalizer.cs | 62 ++ .../Qyl.Cli/Codex/CodexObserverJsonContext.cs | 2 + packages/Qyl.Cli/Codex/CodexObserverModels.cs | 27 + .../Qyl.Cli/Codex/CodexObserverRuntime.cs | 287 ++++++- .../Codex/DiagnosticSnapshotCapture.cs | 566 +++++++++++++ .../Qyl.Cli/Codex/DiagnosticSnapshotInbox.cs | 331 ++++++++ .../Qyl.Cli/Codex/ObserverBridgeServer.cs | 414 ++++++++- packages/Qyl.Cli/Codex/WorkflowJournalPump.cs | 20 +- packages/Qyl.Cli/Codex/WorkflowSpool.cs | 23 +- .../Qyl.Cli/Codex/WorkflowSpoolProtector.cs | 6 + .../Codex/WorkflowTelemetryProjection.cs | 103 ++- .../CollectorSemanticAttributeCatalog.g.cs | 14 + .../Qyl.Cli.Tests/DiagnosticSnapshotTests.cs | 786 ++++++++++++++++++ .../AiDiagnosticSpanEventPersistenceTests.cs | 130 +++ 16 files changed, 2727 insertions(+), 63 deletions(-) create mode 100644 packages/Qyl.Cli/Codex/DiagnosticSnapshotCapture.cs create mode 100644 packages/Qyl.Cli/Codex/DiagnosticSnapshotInbox.cs create mode 100644 tests/Qyl.Cli.Tests/DiagnosticSnapshotTests.cs create mode 100644 tests/Qyl.Collector.Tests/AiDiagnosticSpanEventPersistenceTests.cs diff --git a/eng/config/collector-semantic-policy.json b/eng/config/collector-semantic-policy.json index 647a34ff..04f7e87a 100644 --- a/eng/config/collector-semantic-policy.json +++ b/eng/config/collector-semantic-policy.json @@ -45,6 +45,15 @@ }, "developmentAttributeAllowList": { "span": [ + "qyl.agent.diagnostic.check.count", + "qyl.agent.diagnostic.check.failed_count", + "qyl.agent.diagnostic.extension.id", + "qyl.agent.diagnostic.format.version", + "qyl.agent.diagnostic.outcome", + "qyl.agent.diagnostic.phase", + "qyl.agent.diagnostic.probe.id", + "qyl.agent.diagnostic.snapshot.id", + "qyl.agent.diagnostic.variable.count", "qyl.exception.source", "qyl.instrumentation.domain", "qyl.mcp.evaluation_run.id", @@ -54,7 +63,12 @@ "qyl.mcp.sdk.tier", "qyl.mcp.server.id", "qyl.mcp.test_case.id", - "qyl.mcp.tool.name" + "qyl.mcp.tool.name", + "qyl.workflow.agent.id", + "qyl.workflow.attempt.id", + "qyl.workflow.event.id", + "qyl.workflow.run.id", + "qyl.workflow.tool_call.id" ], "log": [ "browser.device_memory", diff --git a/packages/Qyl.Cli/Codex/ActiveWorkflowRunStore.cs b/packages/Qyl.Cli/Codex/ActiveWorkflowRunStore.cs index 1a00c15c..2ee426d3 100644 --- a/packages/Qyl.Cli/Codex/ActiveWorkflowRunStore.cs +++ b/packages/Qyl.Cli/Codex/ActiveWorkflowRunStore.cs @@ -14,10 +14,13 @@ internal sealed class ActiveWorkflowRunStore public ActiveWorkflowRunStore(string root) { Directory.CreateDirectory(root); + Root = root; _activePath = Path.Combine(root, ActiveFileName); _lockPath = Path.Combine(root, LockFileName); } + public string Root { get; } + public FileStream Acquire() { try diff --git a/packages/Qyl.Cli/Codex/CodexEventNormalizer.cs b/packages/Qyl.Cli/Codex/CodexEventNormalizer.cs index efd42ab6..dc3e47ff 100644 --- a/packages/Qyl.Cli/Codex/CodexEventNormalizer.cs +++ b/packages/Qyl.Cli/Codex/CodexEventNormalizer.cs @@ -10,6 +10,8 @@ internal sealed class CodexEventNormalizer private readonly HashSet _eventIds = new(StringComparer.Ordinal); private readonly Dictionary _threads = new(StringComparer.Ordinal); private readonly Dictionary _approvals = new(StringComparer.Ordinal); + private readonly Dictionary _diagnosticSnapshots = + new(StringComparer.Ordinal); private ulong _sourceSequence; private string? _rootThreadId; private string? _activeRootTurnId; @@ -119,6 +121,58 @@ public CodexNormalizedBatch CompleteRun(DateTimeOffset timestamp, bool succeeded : new CodexNormalizedBatch([workflowEvent], []); } + public CodexNormalizedBatch NormalizeDiagnosticSnapshot( + DiagnosticSnapshotInboxRequest request, + DateTimeOffset receivedAt) + { + if (_diagnosticSnapshots.TryGetValue(request.SnapshotId, out var previous)) + { + if (!string.Equals(previous.PayloadDigest, request.PayloadDigest, StringComparison.Ordinal)) + throw new DiagnosticSnapshotConflictException(); + return previous.Batch ?? default; + } + if (_rootThreadId is null || !_threads.TryGetValue(_rootThreadId, out var root)) + throw new DiagnosticSnapshotContextUnavailableException(); + + var eventId = StableEventId("diagnostic", request.SnapshotId); + var workflowEvent = CreateEvent( + eventId, + WorkflowJournalEventKind.ContentCaptured, + receivedAt, + _rootThreadId, + root.ActiveTurnId, + root.AttemptId, + null, + null, + null, + null, + [request.Content.ContentRef], + new Dictionary(StringComparer.Ordinal) + { + ["extension_id"] = DiagnosticSnapshotCapture.ExtensionId, + ["format_version"] = DiagnosticSnapshotCapture.FormatVersion, + ["snapshot_id"] = request.SnapshotId, + ["probe_id"] = request.ProbeId, + ["phase"] = request.Phase, + ["outcome"] = request.Outcome, + ["variable_count"] = request.VariableCount, + ["check_count"] = request.CheckCount, + ["failed_check_count"] = request.FailedCheckCount, + ["content_ref"] = request.Content.ContentRef.Value + }) ?? throw new DiagnosticSnapshotConflictException(); + var batch = new CodexNormalizedBatch([workflowEvent], [request.Content]); + _diagnosticSnapshots.Add( + request.SnapshotId, + new DiagnosticSnapshotContext(request.PayloadDigest, batch)); + return batch; + } + + public void MarkDiagnosticSnapshotRecorded(string snapshotId) + { + if (_diagnosticSnapshots.TryGetValue(snapshotId, out var context)) + _diagnosticSnapshots[snapshotId] = context with { Batch = null }; + } + private void NormalizeThreadStarted( JsonElement parameters, DateTimeOffset receivedAt, @@ -1008,4 +1062,12 @@ private sealed record ApprovalContext( string TurnId, string? ItemId, string? AttemptId); + + private sealed record DiagnosticSnapshotContext( + string PayloadDigest, + CodexNormalizedBatch? Batch); } + +internal sealed class DiagnosticSnapshotConflictException : Exception; + +internal sealed class DiagnosticSnapshotContextUnavailableException : Exception; diff --git a/packages/Qyl.Cli/Codex/CodexObserverJsonContext.cs b/packages/Qyl.Cli/Codex/CodexObserverJsonContext.cs index 46dd5bd6..7a7853cc 100644 --- a/packages/Qyl.Cli/Codex/CodexObserverJsonContext.cs +++ b/packages/Qyl.Cli/Codex/CodexObserverJsonContext.cs @@ -32,5 +32,7 @@ internal partial class CodexWorkflowContractJsonContext : JsonSerializerContext; [JsonSerializable(typeof(WorkflowSpoolEntry[]))] [JsonSerializable(typeof(WorkflowSpoolEnvelope))] [JsonSerializable(typeof(ActiveWorkflowRun))] +[JsonSerializable(typeof(DiagnosticSnapshotInboxRequest))] +[JsonSerializable(typeof(DiagnosticSnapshotInboxAcknowledgement))] [JsonSerializable(typeof(string))] internal partial class CodexObserverStateJsonContext : JsonSerializerContext; diff --git a/packages/Qyl.Cli/Codex/CodexObserverModels.cs b/packages/Qyl.Cli/Codex/CodexObserverModels.cs index 7beeff5c..c445721d 100644 --- a/packages/Qyl.Cli/Codex/CodexObserverModels.cs +++ b/packages/Qyl.Cli/Codex/CodexObserverModels.cs @@ -27,6 +27,33 @@ internal sealed record ActiveWorkflowRun( DateTimeOffset StartedAt, int ProcessId); +internal sealed record DiagnosticSnapshotInboxRequest( + string RunId, + string SnapshotId, + string ProbeId, + string Phase, + string Outcome, + int VariableCount, + int CheckCount, + int FailedCheckCount, + string PayloadDigest, + DateTimeOffset SubmittedAt, + Qyl.Api.Contracts.Workflow.WorkflowContentChunk Content); + +internal sealed record DiagnosticSnapshotInboxAcknowledgement( + string RunId, + string SnapshotId, + string PayloadDigest, + string Status, + string Code, + string? EventId); + +internal readonly record struct DiagnosticSnapshotSubmissionResult( + bool Recorded, + string Code, + string SnapshotId, + string? EventId); + internal sealed record CodexSchemaIdentity( string CodexVersion, string SchemaDirectory, diff --git a/packages/Qyl.Cli/Codex/CodexObserverRuntime.cs b/packages/Qyl.Cli/Codex/CodexObserverRuntime.cs index c087d486..cecb43e3 100644 --- a/packages/Qyl.Cli/Codex/CodexObserverRuntime.cs +++ b/packages/Qyl.Cli/Codex/CodexObserverRuntime.cs @@ -2,6 +2,7 @@ using System.Net; using System.Net.Sockets; using System.Security.Cryptography; +using System.Text.Json; namespace Qyl.Cli.Codex; @@ -28,6 +29,7 @@ public static async Task RunAsync( await using var activeLockScope = activeLock.ConfigureAwait(false); using var shutdown = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + using var diagnosticShutdown = CancellationTokenSource.CreateLinkedTokenSource(shutdown.Token); ConsoleCancelEventHandler cancelHandler = (_, eventArgs) => { eventArgs.Cancel = true; @@ -46,11 +48,18 @@ public static async Task RunAsync( CodexAppServerClient? observer = null; Task? uploadTask = null; Task? controlTask = null; + Task? diagnosticTask = null; WorkflowSpool? spool = null; WorkflowSpoolMetadata? metadata = null; + DiagnosticSnapshotInbox? diagnosticInbox = null; + var diagnosticsPrepared = false; + var diagnosticsStopped = false; + using var journalGate = new SemaphoreSlim(1, 1); var normalizer = new CodexEventNormalizer(); + var acceptingObserverMessages = true; var runCompleted = false; using var telemetry = WorkflowTelemetryProjection.Create( + runId, Environment.GetEnvironmentVariable(ApiKeyEnvironment)); using var httpClient = new HttpClient { @@ -65,6 +74,9 @@ public static async Task RunAsync( var spoolStore = new WorkflowSpoolStore(root); spool = spoolStore.Open(runId); + diagnosticInbox = new DiagnosticSnapshotInbox(root); + diagnosticInbox.PrepareRun(runId); + diagnosticsPrepared = true; metadata = new WorkflowSpoolMetadata( runId, null, @@ -79,10 +91,21 @@ await activeRuns.WriteAsync( new ActiveWorkflowRun(runId, null, startedAt, Environment.ProcessId), shutdown.Token).ConfigureAwait(false); - await AppendBatchAsync( + var startBatch = normalizer.StartRun(startedAt); + await AppendBatchAsync(spool, startBatch, shutdown.Token).ConfigureAwait(false); + foreach (var workflowEvent in startBatch.Events ?? []) + telemetry.Record(workflowEvent); + +#pragma warning disable CA2025 // Awaited during shutdown before journalGate leaves scope. + diagnosticTask = RunDiagnosticDrainLoopAsync( + diagnosticInbox, + runId, + normalizer, spool, - normalizer.StartRun(startedAt), - shutdown.Token).ConfigureAwait(false); + telemetry.Record, + journalGate, + diagnosticShutdown.Token); +#pragma warning restore CA2025 var token = Base64Url(RandomNumberGenerator.GetBytes(32)); var tokenPath = Path.Combine(runtimeDirectory, "capability-token"); @@ -103,32 +126,42 @@ await AppendBatchAsync( TaskCreationOptions.RunContinuationsAsynchronously); observer.MessageReceived += async message => { - var batch = normalizer.Normalize(message, TimeProvider.System.GetUtcNow()); - if (batch.Events is null || batch.Events.Count is 0) - return; - - if (normalizer.RootThreadId is not null && - metadata.ThreadId != normalizer.RootThreadId) + await journalGate.WaitAsync(shutdown.Token).ConfigureAwait(false); + try { - metadata = metadata with + if (!acceptingObserverMessages) + return; + var batch = normalizer.Normalize(message, TimeProvider.System.GetUtcNow()); + if (batch.Events is null || batch.Events.Count is 0) + return; + + if (normalizer.RootThreadId is not null && + metadata.ThreadId != normalizer.RootThreadId) { - ThreadId = normalizer.RootThreadId, - Title = normalizer.RootTitle - }; - await spool.WriteMetadataAsync(metadata, shutdown.Token).ConfigureAwait(false); - await activeRuns.WriteAsync( - new ActiveWorkflowRun( - runId, - normalizer.RootThreadId, - startedAt, - Environment.ProcessId), - shutdown.Token).ConfigureAwait(false); - rootThreadReady.TrySetResult(); + metadata = metadata with + { + ThreadId = normalizer.RootThreadId, + Title = normalizer.RootTitle + }; + await spool.WriteMetadataAsync(metadata, shutdown.Token).ConfigureAwait(false); + await activeRuns.WriteAsync( + new ActiveWorkflowRun( + runId, + normalizer.RootThreadId, + startedAt, + Environment.ProcessId), + shutdown.Token).ConfigureAwait(false); + rootThreadReady.TrySetResult(); + } + + foreach (var workflowEvent in batch.Events) + telemetry.Record(workflowEvent); + await AppendBatchAsync(spool, batch, shutdown.Token).ConfigureAwait(false); + } + finally + { + journalGate.Release(); } - - foreach (var workflowEvent in batch.Events) - telemetry.Record(workflowEvent); - await AppendBatchAsync(spool, batch, shutdown.Token).ConfigureAwait(false); }; await observer.ConnectAsync(endpoint, token, shutdown.Token).ConfigureAwait(false); @@ -138,13 +171,16 @@ await activeRuns.WriteAsync( Environment.GetEnvironmentVariable(ApiKeyEnvironment)); var pump = new WorkflowJournalPump(spoolStore, collector); uploadTask = pump.RunUploadLoopAsync(shutdown.Token); +#pragma warning disable CA2025 // Awaited during shutdown before journalGate leaves scope. controlTask = RunControlsWhenReadyAsync( rootThreadReady.Task, runId, pump, normalizer, observer, + journalGate, shutdown.Token); +#pragma warning restore CA2025 tui = StartTui(codexExecutable, endpoint, token, codexArguments); var tuiExit = tui.WaitForExitAsync(shutdown.Token); @@ -166,14 +202,37 @@ await activeRuns.WriteAsync( } await tuiExit.ConfigureAwait(false); - var finalBatch = normalizer.CompleteRun( - TimeProvider.System.GetUtcNow(), - tui.ExitCode is 0); - foreach (var workflowEvent in finalBatch.Events ?? []) - telemetry.Record(workflowEvent); - await AppendBatchAsync(spool, finalBatch, CancellationToken.None).ConfigureAwait(false); - metadata = metadata with { Sealed = true }; - await spool.WriteMetadataAsync(metadata, CancellationToken.None).ConfigureAwait(false); + activeRuns.Clear(runId); + diagnosticInbox.CloseRun(runId); + await StopDiagnosticDrainAsync( + diagnosticShutdown, + diagnosticTask, + diagnosticInbox, + runId, + normalizer, + spool, + telemetry.Record, + journalGate).ConfigureAwait(false); + diagnosticsStopped = true; + + await journalGate.WaitAsync(CancellationToken.None).ConfigureAwait(false); + try + { + acceptingObserverMessages = false; + activeRuns.Clear(runId); + var finalBatch = normalizer.CompleteRun( + TimeProvider.System.GetUtcNow(), + tui.ExitCode is 0); + foreach (var workflowEvent in finalBatch.Events ?? []) + telemetry.Record(workflowEvent); + await AppendBatchAsync(spool, finalBatch, CancellationToken.None).ConfigureAwait(false); + metadata = metadata with { Sealed = true }; + await spool.WriteMetadataAsync(metadata, CancellationToken.None).ConfigureAwait(false); + } + finally + { + journalGate.Release(); + } runCompleted = true; using var flushTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); @@ -194,14 +253,39 @@ await activeRuns.WriteAsync( } finally { + activeRuns.Clear(runId); + if (!diagnosticsStopped && diagnosticsPrepared) + { + diagnosticInbox!.CloseRun(runId); + await StopDiagnosticDrainAsync( + diagnosticShutdown, + diagnosticTask, + diagnosticInbox, + runId, + normalizer, + spool, + telemetry.Record, + journalGate).ConfigureAwait(false); + } + if (!runCompleted && spool is not null && metadata is not null) { - var failedBatch = normalizer.CompleteRun(TimeProvider.System.GetUtcNow(), succeeded: false); - foreach (var workflowEvent in failedBatch.Events ?? []) - telemetry.Record(workflowEvent); - await AppendBatchAsync(spool, failedBatch, CancellationToken.None).ConfigureAwait(false); - metadata = metadata with { Sealed = true }; - await spool.WriteMetadataAsync(metadata, CancellationToken.None).ConfigureAwait(false); + await journalGate.WaitAsync(CancellationToken.None).ConfigureAwait(false); + try + { + acceptingObserverMessages = false; + activeRuns.Clear(runId); + var failedBatch = normalizer.CompleteRun(TimeProvider.System.GetUtcNow(), succeeded: false); + foreach (var workflowEvent in failedBatch.Events ?? []) + telemetry.Record(workflowEvent); + await AppendBatchAsync(spool, failedBatch, CancellationToken.None).ConfigureAwait(false); + metadata = metadata with { Sealed = true }; + await spool.WriteMetadataAsync(metadata, CancellationToken.None).ConfigureAwait(false); + } + finally + { + journalGate.Release(); + } } await shutdown.CancelAsync().ConfigureAwait(false); @@ -216,7 +300,6 @@ await activeRuns.WriteAsync( await DisposeObserverAsync(observer).ConfigureAwait(false); if (appServer is not null) await appServer.DisposeAsync().ConfigureAwait(false); - activeRuns.Clear(runId); Console.CancelKeyPress -= cancelHandler; DeleteRuntimeDirectory(runtimeDirectory); } @@ -241,6 +324,7 @@ private static async Task RunControlsWhenReadyAsync( WorkflowJournalPump pump, CodexEventNormalizer normalizer, CodexAppServerClient observer, + SemaphoreSlim journalGate, CancellationToken cancellationToken) { await rootThreadReady.WaitAsync(cancellationToken).ConfigureAwait(false); @@ -248,9 +332,130 @@ await pump.RunControlLoopAsync( runId, normalizer, observer, + journalGate, cancellationToken).ConfigureAwait(false); } + private static async Task RunDiagnosticDrainLoopAsync( + DiagnosticSnapshotInbox inbox, + string runId, + CodexEventNormalizer normalizer, + WorkflowSpool spool, + Action recordTelemetry, + SemaphoreSlim journalGate, + CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + await DrainDiagnosticsOnceAsync( + inbox, + runId, + normalizer, + spool, + recordTelemetry, + journalGate, + cancellationToken) + .ConfigureAwait(false); + } + catch (Exception exception) when ( + exception is IOException or InvalidDataException or CryptographicException or JsonException) + { + Console.Error.WriteLine( + $"[qyl] Diagnostic inbox drain failed (diagnostic_inbox_unreadable): {exception.GetType().Name}"); + } + await Task.Delay(50, cancellationToken).ConfigureAwait(false); + } + } + + private static async Task StopDiagnosticDrainAsync( + CancellationTokenSource diagnosticShutdown, + Task? diagnosticTask, + DiagnosticSnapshotInbox inbox, + string runId, + CodexEventNormalizer normalizer, + WorkflowSpool? spool, + Action recordTelemetry, + SemaphoreSlim journalGate) + { + await diagnosticShutdown.CancelAsync().ConfigureAwait(false); + if (diagnosticTask is not null) + await IgnoreExpectedShutdownAsync(diagnosticTask).ConfigureAwait(false); + if (spool is not null) + { + await DrainDiagnosticsOnceAsync( + inbox, + runId, + normalizer, + spool, + recordTelemetry, + journalGate, + CancellationToken.None) + .ConfigureAwait(false); + } + } + + internal static async Task DrainDiagnosticsOnceAsync( + DiagnosticSnapshotInbox inbox, + string runId, + CodexEventNormalizer normalizer, + WorkflowSpool spool, + Action? recordTelemetry, + SemaphoreSlim journalGate, + CancellationToken cancellationToken) + { + var recorded = 0; + foreach (var request in inbox.ReadPending(runId)) + { + CodexNormalizedBatch batch; + await journalGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + batch = normalizer.NormalizeDiagnosticSnapshot( + request, + TimeProvider.System.GetUtcNow()); + if (batch.Events is { Count: > 0 }) + { + await AppendBatchAsync(spool, batch, CancellationToken.None).ConfigureAwait(false); + normalizer.MarkDiagnosticSnapshotRecorded(request.SnapshotId); + foreach (var workflowEvent in batch.Events) + recordTelemetry?.Invoke(workflowEvent); + } + } + catch (DiagnosticSnapshotContextUnavailableException) + { + continue; + } + catch (DiagnosticSnapshotConflictException) + { + await inbox.AcknowledgeAsync( + request, + "failed", + "snapshot_conflict", + null, + cancellationToken).ConfigureAwait(false); + continue; + } + finally + { + journalGate.Release(); + } + + var eventId = batch.Events is { Count: > 0 } + ? batch.Events[0].EventId.Value + : $"diagnostic:{request.SnapshotId}"; + await inbox.AcknowledgeAsync( + request, + "recorded", + "recorded", + eventId, + cancellationToken).ConfigureAwait(false); + recorded++; + } + return recorded; + } + private static async Task AppendBatchAsync( WorkflowSpool spool, CodexNormalizedBatch batch, diff --git a/packages/Qyl.Cli/Codex/DiagnosticSnapshotCapture.cs b/packages/Qyl.Cli/Codex/DiagnosticSnapshotCapture.cs new file mode 100644 index 00000000..60d13865 --- /dev/null +++ b/packages/Qyl.Cli/Codex/DiagnosticSnapshotCapture.cs @@ -0,0 +1,566 @@ +using System.Buffers; +using System.Globalization; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using Qyl.Api.Contracts.Workflow; + +namespace Qyl.Cli.Codex; + +internal static class DiagnosticSnapshotCapture +{ + internal const string ExtensionId = "qyl.agent.diagnostic.snapshot"; + internal const int FormatVersion = 1; + internal const int MaxIdentifierLength = 128; + internal const int MaxVariables = 64; + internal const int MaxChecks = 64; + internal const int MaxValueDepth = 8; + internal const int MaxValueBytes = 16 * 1024; + internal const int MaxInputBytes = 192 * 1024; + internal const int MaxCapturedBytes = 64 * 1024; + + private static readonly HashSet s_phases = + ["input", "output", "error", "checkpoint"]; + private static readonly HashSet s_classifications = + ["public", "internal", "sensitive", "secret"]; + private static readonly HashSet s_operators = + [ + "equal", + "not_equal", + "exists", + "type_is", + "contains", + "less_than", + "greater_than" + ]; + private static readonly HashSet s_valueTypes = + ["null", "boolean", "integer", "number", "string", "json"]; + + public static bool TryCreate( + ActiveWorkflowRun active, + JsonElement arguments, + WorkflowSpoolProtector protector, + DateTimeOffset submittedAt, + out DiagnosticSnapshotInboxRequest? request, + out DiagnosticSnapshotValidationError error) + { + request = null; + error = default; + + if (arguments.ValueKind is not JsonValueKind.Object) + return Fail("invalid_input", "arguments", out error); + if (Encoding.UTF8.GetByteCount(arguments.GetRawText()) > MaxInputBytes) + return Fail("payload_too_large", "arguments", out error); + if (!HasOnlyProperties( + arguments, + ["snapshotId", "probeId", "phase", "variables", "checks"], + out var invalidRootProperty)) + { + return Fail("invalid_property", invalidRootProperty, out error); + } + if (!TryIdentifier(arguments, "snapshotId", out var snapshotId)) + return Fail("invalid_snapshot_id", "snapshotId", out error); + if (!TryIdentifier(arguments, "probeId", out var probeId)) + return Fail("invalid_probe_id", "probeId", out error); + if (!TryRequiredString(arguments, "phase", out var phase) || !s_phases.Contains(phase)) + return Fail("invalid_phase", "phase", out error); + if (!arguments.TryGetProperty("variables", out var variablesElement) || + variablesElement.ValueKind is not JsonValueKind.Array) + { + return Fail("invalid_variables", "variables", out error); + } + if (variablesElement.GetArrayLength() > MaxVariables) + return Fail("too_many_variables", "variables", out error); + + var variables = new List(variablesElement.GetArrayLength()); + var variablesByName = new Dictionary(StringComparer.Ordinal); + var variableIndex = 0; + foreach (var variableElement in variablesElement.EnumerateArray()) + { + var path = $"variables[{variableIndex.ToString(CultureInfo.InvariantCulture)}]"; + variableIndex++; + if (variableElement.ValueKind is not JsonValueKind.Object || + !HasOnlyProperties(variableElement, ["name", "classification", "value"], out _)) + { + return Fail("invalid_variable", path, out error); + } + if (!TryIdentifier(variableElement, "name", out var name)) + return Fail("invalid_variable_name", $"{path}.name", out error); + if (!TryRequiredString(variableElement, "classification", out var classification) || + !s_classifications.Contains(classification)) + { + return Fail("invalid_classification", $"{path}.classification", out error); + } + if (!variableElement.TryGetProperty("value", out var value)) + return Fail("missing_value", $"{path}.value", out error); + if (JsonDepth(value) > MaxValueDepth) + return Fail("value_too_deep", $"{path}.value", out error); + byte[] canonicalValue; + try + { + canonicalValue = CanonicalJson(value); + } + catch (InvalidDataException) + { + return Fail("invalid_json_value", $"{path}.value", out error); + } + if (canonicalValue.Length > MaxValueBytes) + return Fail("value_too_large", $"{path}.value", out error); + + var variable = new DiagnosticVariable( + name, + classification, + ValueType(value), + value, + canonicalValue); + if (!variablesByName.TryAdd(name, variable)) + return Fail("duplicate_variable", $"{path}.name", out error); + variables.Add(variable); + } + + var checks = new List(); + if (arguments.TryGetProperty("checks", out var checksElement)) + { + if (checksElement.ValueKind is not JsonValueKind.Array) + return Fail("invalid_checks", "checks", out error); + if (checksElement.GetArrayLength() > MaxChecks) + return Fail("too_many_checks", "checks", out error); + + var checkIds = new HashSet(StringComparer.Ordinal); + var checkIndex = 0; + foreach (var checkElement in checksElement.EnumerateArray()) + { + var path = $"checks[{checkIndex.ToString(CultureInfo.InvariantCulture)}]"; + checkIndex++; + if (checkElement.ValueKind is not JsonValueKind.Object || + !HasOnlyProperties( + checkElement, + ["checkId", "operator", "actual", "expected", "expectedType"], + out _)) + { + return Fail("invalid_check", path, out error); + } + if (!TryIdentifier(checkElement, "checkId", out var checkId)) + return Fail("invalid_check_id", $"{path}.checkId", out error); + if (!checkIds.Add(checkId)) + return Fail("duplicate_check", $"{path}.checkId", out error); + if (!TryRequiredString(checkElement, "operator", out var checkOperator) || + !s_operators.Contains(checkOperator)) + { + return Fail("invalid_operator", $"{path}.operator", out error); + } + if (!TryIdentifier(checkElement, "actual", out var actualName)) + { + return Fail("invalid_actual_variable", $"{path}.actual", out error); + } + variablesByName.TryGetValue(actualName, out var actualVariable); + + var hasExpected = checkElement.TryGetProperty("expected", out var expectedElement); + var hasExpectedType = checkElement.TryGetProperty("expectedType", out var expectedTypeElement); + string? expectedName = null; + string? expectedType = null; + DiagnosticVariable? expectedVariable = null; + if (checkOperator is "equal" or "not_equal" or "contains" or "less_than" or "greater_than") + { + if (!hasExpected || hasExpectedType || + expectedElement.ValueKind is not JsonValueKind.String || + !IsMachineIdentifier(expectedElement.GetString(), out expectedName)) + { + return Fail("invalid_expected_variable", $"{path}.expected", out error); + } + variablesByName.TryGetValue(expectedName, out expectedVariable); + } + else if (checkOperator == "type_is") + { + if (hasExpected || !hasExpectedType || + expectedTypeElement.ValueKind is not JsonValueKind.String || + (expectedType = expectedTypeElement.GetString()) is null || + !s_valueTypes.Contains(expectedType)) + { + return Fail("invalid_expected_type", $"{path}.expectedType", out error); + } + } + else if (hasExpected || hasExpectedType) + { + return Fail("unexpected_check_operand", path, out error); + } + + var checkOutcome = Evaluate( + checkOperator, + actualVariable, + expectedVariable, + expectedType); + checks.Add(new DiagnosticCheck( + checkId, + checkOperator, + actualName, + expectedName, + expectedType, + checkOutcome)); + } + } + + var failedChecks = checks.Count(static check => check.Outcome == "fail"); + var unknownChecks = checks.Count(static check => check.Outcome == "unknown"); + var outcome = checks.Count == 0 + ? "not_evaluated" + : failedChecks > 0 + ? "fail" + : unknownChecks > 0 + ? "unknown" + : "pass"; + + var semanticPayload = WritePayload( + snapshotId, + probeId, + phase, + outcome, + variables, + checks, + captureNonce: null, + redact: false); + var payloadDigest = protector.KeyedDigest(semanticPayload); + CryptographicOperations.ZeroMemory(semanticPayload); + var captureNonce = RandomNumberGenerator.GetBytes(16); + var capturedPayload = WritePayload( + snapshotId, + probeId, + phase, + outcome, + variables, + checks, + Convert.ToHexStringLower(captureNonce), + redact: true); + foreach (var variable in variables) + CryptographicOperations.ZeroMemory(variable.CanonicalValue); + if (capturedPayload.Length > MaxCapturedBytes) + return Fail("payload_too_large", "arguments", out error); + + var contentHash = Convert.ToHexStringLower(SHA256.HashData(capturedPayload)); + var chunk = new WorkflowContentChunk + { + ContentRef = new WorkflowContentRef($"sha256:{contentHash}"), + ContentType = "application/json", + Encoding = WorkflowContentEncoding.Utf8, + Content = Encoding.UTF8.GetString(capturedPayload) + }; + + request = new DiagnosticSnapshotInboxRequest( + active.RunId, + snapshotId, + probeId, + phase, + outcome, + variables.Count, + checks.Count, + failedChecks, + payloadDigest, + submittedAt, + chunk); + return true; + } + + private static string Evaluate( + string checkOperator, + DiagnosticVariable? actual, + DiagnosticVariable? expected, + string? expectedType) + { + if (checkOperator == "exists") + return actual is not null && actual.Value.ValueKind is not JsonValueKind.Null ? "pass" : "fail"; + if (actual is null) + return "unknown"; + return checkOperator switch + { + "type_is" => actual.ValueType == expectedType ? "pass" : "fail", + "equal" => Equality(actual, expected, negate: false), + "not_equal" => Equality(actual, expected, negate: true), + "contains" => expected is null ? "unknown" : Contains(actual.Value, expected.Value), + "less_than" => expected is null + ? "unknown" + : CompareNumbers(actual.Value, expected.Value, lessThan: true), + "greater_than" => expected is null + ? "unknown" + : CompareNumbers(actual.Value, expected.Value, lessThan: false), + _ => "unknown" + }; + } + + private static string Equality( + DiagnosticVariable actual, + DiagnosticVariable? expected, + bool negate) + { + if (expected is null) + return "unknown"; + var bothNumeric = actual.ValueType is "integer" or "number" && + expected.ValueType is "integer" or "number"; + if (actual.ValueType != expected.ValueType && !bothNumeric) + return "unknown"; + + bool equal; + if (bothNumeric && + decimal.TryParse(actual.Value.GetRawText(), NumberStyles.Float, CultureInfo.InvariantCulture, out var left) && + decimal.TryParse(expected.Value.GetRawText(), NumberStyles.Float, CultureInfo.InvariantCulture, out var right)) + { + equal = left == right; + } + else if (bothNumeric) + { + return "unknown"; + } + else + { + equal = JsonElement.DeepEquals(actual.Value, expected.Value); + } + return equal != negate ? "pass" : "fail"; + } + + private static string Contains(JsonElement actual, JsonElement expected) + { + if (actual.ValueKind is JsonValueKind.String && expected.ValueKind is JsonValueKind.String) + { + return actual.GetString()!.Contains(expected.GetString()!, StringComparison.Ordinal) + ? "pass" + : "fail"; + } + if (actual.ValueKind is JsonValueKind.Array) + { + return actual.EnumerateArray().Any(item => JsonElement.DeepEquals(item, expected)) + ? "pass" + : "fail"; + } + return "unknown"; + } + + private static string CompareNumbers(JsonElement actual, JsonElement expected, bool lessThan) + { + if (actual.ValueKind is not JsonValueKind.Number || + expected.ValueKind is not JsonValueKind.Number || + !decimal.TryParse(actual.GetRawText(), NumberStyles.Float, CultureInfo.InvariantCulture, out var left) || + !decimal.TryParse(expected.GetRawText(), NumberStyles.Float, CultureInfo.InvariantCulture, out var right)) + { + return "unknown"; + } + return lessThan ? left < right ? "pass" : "fail" : left > right ? "pass" : "fail"; + } + + private static byte[] WritePayload( + string snapshotId, + string probeId, + string phase, + string outcome, + IReadOnlyList variables, + IReadOnlyList checks, + string? captureNonce, + bool redact) + { + var buffer = new ArrayBufferWriter(); + using var writer = new Utf8JsonWriter(buffer); + writer.WriteStartObject(); + writer.WriteString("extension_id", ExtensionId); + writer.WriteNumber("format_version", FormatVersion); + if (captureNonce is not null) + writer.WriteString("capture_nonce", captureNonce); + writer.WriteString("snapshot_id", snapshotId); + writer.WriteString("probe_id", probeId); + writer.WriteString("phase", phase); + writer.WriteString("outcome", outcome); + writer.WritePropertyName("variables"); + writer.WriteStartArray(); + foreach (var variable in variables.OrderBy(static item => item.Name, StringComparer.Ordinal)) + { + writer.WriteStartObject(); + writer.WriteString("name", variable.Name); + writer.WriteString("classification", variable.Classification); + writer.WriteString("type", variable.ValueType); + if (!redact || variable.Classification is "public" or "internal") + { + writer.WriteString("capture", "value"); + writer.WritePropertyName("value"); + writer.WriteRawValue(variable.CanonicalValue, skipInputValidation: true); + } + else + { + writer.WriteString( + "capture", + variable.Classification == "sensitive" ? "redacted" : "omitted"); + } + writer.WriteEndObject(); + } + writer.WriteEndArray(); + writer.WritePropertyName("checks"); + writer.WriteStartArray(); + foreach (var check in checks.OrderBy(static item => item.CheckId, StringComparer.Ordinal)) + { + writer.WriteStartObject(); + writer.WriteString("check_id", check.CheckId); + writer.WriteString("operator", check.Operator); + writer.WriteString("actual", check.Actual); + if (check.Expected is not null) + writer.WriteString("expected", check.Expected); + if (check.ExpectedType is not null) + writer.WriteString("expected_type", check.ExpectedType); + writer.WriteString("outcome", check.Outcome); + writer.WriteEndObject(); + } + writer.WriteEndArray(); + writer.WriteEndObject(); + writer.Flush(); + return buffer.WrittenSpan.ToArray(); + } + + private static byte[] CanonicalJson(JsonElement value) + { + var buffer = new ArrayBufferWriter(); + using var writer = new Utf8JsonWriter(buffer); + WriteCanonicalValue(writer, value); + writer.Flush(); + return buffer.WrittenSpan.ToArray(); + } + + private static void WriteCanonicalValue(Utf8JsonWriter writer, JsonElement value) + { + switch (value.ValueKind) + { + case JsonValueKind.Object: + writer.WriteStartObject(); + var properties = value.EnumerateObject().ToArray(); + var names = new HashSet(StringComparer.Ordinal); + foreach (var property in properties.OrderBy(static item => item.Name, StringComparer.Ordinal)) + { + if (!names.Add(property.Name)) + throw new InvalidDataException("Diagnostic JSON values cannot contain duplicate properties."); + writer.WritePropertyName(property.Name); + WriteCanonicalValue(writer, property.Value); + } + writer.WriteEndObject(); + break; + case JsonValueKind.Array: + writer.WriteStartArray(); + foreach (var item in value.EnumerateArray()) + WriteCanonicalValue(writer, item); + writer.WriteEndArray(); + break; + case JsonValueKind.String: + writer.WriteStringValue(value.GetString()); + break; + case JsonValueKind.Number: + writer.WriteRawValue(value.GetRawText(), skipInputValidation: true); + break; + case JsonValueKind.True: + writer.WriteBooleanValue(true); + break; + case JsonValueKind.False: + writer.WriteBooleanValue(false); + break; + case JsonValueKind.Null: + writer.WriteNullValue(); + break; + default: + throw new InvalidDataException("Diagnostic values must be closed JSON values."); + } + } + + private static int JsonDepth(JsonElement value) + { + return value.ValueKind switch + { + JsonValueKind.Object => 1 + value.EnumerateObject() + .Select(static property => JsonDepth(property.Value)) + .DefaultIfEmpty(0) + .Max(), + JsonValueKind.Array => 1 + value.EnumerateArray() + .Select(static item => JsonDepth(item)) + .DefaultIfEmpty(0) + .Max(), + _ => 0 + }; + } + + private static string ValueType(JsonElement value) => value.ValueKind switch + { + JsonValueKind.Null => "null", + JsonValueKind.True or JsonValueKind.False => "boolean", + JsonValueKind.Number when value.TryGetInt64(out _) => "integer", + JsonValueKind.Number => "number", + JsonValueKind.String => "string", + JsonValueKind.Object or JsonValueKind.Array => "json", + _ => throw new InvalidDataException("Diagnostic values must be closed JSON values.") + }; + + private static bool HasOnlyProperties( + JsonElement element, + string[] allowed, + out string invalidProperty) + { + invalidProperty = ""; + var seen = new HashSet(StringComparer.Ordinal); + foreach (var property in element.EnumerateObject()) + { + if (!allowed.Contains(property.Name) || !seen.Add(property.Name)) + { + invalidProperty = property.Name; + return false; + } + } + return true; + } + + private static bool TryIdentifier(JsonElement value, string name, out string result) + { + result = ""; + return TryRequiredString(value, name, out var candidate) && + IsMachineIdentifier(candidate, out result); + } + + private static bool IsMachineIdentifier(string? candidate, out string result) + { + result = candidate ?? ""; + if (candidate is null || candidate.Length is 0 or > MaxIdentifierLength || + !(IsAsciiLetterOrDigit(candidate[0]) || candidate[0] == '_')) + { + return false; + } + return candidate.All(static character => + IsAsciiLetterOrDigit(character) || + character is '.' or '_' or ':' or '/' or '-' or '[' or ']'); + } + + private static bool IsAsciiLetterOrDigit(char value) => + value is >= 'a' and <= 'z' or >= 'A' and <= 'Z' or >= '0' and <= '9'; + + private static bool TryRequiredString(JsonElement value, string name, out string result) + { + result = ""; + return value.TryGetProperty(name, out var property) && + property.ValueKind is JsonValueKind.String && + (result = property.GetString()!) is not null; + } + + private static bool Fail( + string code, + string field, + out DiagnosticSnapshotValidationError error) + { + error = new DiagnosticSnapshotValidationError(code, field); + return false; + } + + private sealed record DiagnosticVariable( + string Name, + string Classification, + string ValueType, + JsonElement Value, + byte[] CanonicalValue); + + private sealed record DiagnosticCheck( + string CheckId, + string Operator, + string Actual, + string? Expected, + string? ExpectedType, + string Outcome); +} + +internal readonly record struct DiagnosticSnapshotValidationError(string Code, string Field); diff --git a/packages/Qyl.Cli/Codex/DiagnosticSnapshotInbox.cs b/packages/Qyl.Cli/Codex/DiagnosticSnapshotInbox.cs new file mode 100644 index 00000000..5eba51c8 --- /dev/null +++ b/packages/Qyl.Cli/Codex/DiagnosticSnapshotInbox.cs @@ -0,0 +1,331 @@ +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; + +namespace Qyl.Cli.Codex; + +internal sealed class DiagnosticSnapshotInbox +{ + private const string RequestSuffix = ".request.qyl"; + private const string AcknowledgementSuffix = ".ack.qyl"; + private const string LockFileName = "inbox.lock"; + private const string CurrentRunFileName = "current-run"; + private static readonly TimeSpan s_pollInterval = TimeSpan.FromMilliseconds(25); + + private readonly string _inboxDirectory; + private readonly WorkflowSpoolProtector _protector; + + public DiagnosticSnapshotInbox(string root) + { + _inboxDirectory = Path.Combine(root, "diagnostic-inbox"); + Directory.CreateDirectory(_inboxDirectory); + RestrictDirectory(_inboxDirectory); + _protector = WorkflowSpoolProtector.Open(root); + } + + public WorkflowSpoolProtector Protector => _protector; + + public void PrepareRun(string runId) + { + using var inboxLock = AcquireLock(); + foreach (var pattern in new[] + { + $"*{RequestSuffix}", + $"*{AcknowledgementSuffix}", + $"*{RequestSuffix}.corrupt" + }) + { + foreach (var path in Directory.EnumerateFiles( + _inboxDirectory, + pattern, + SearchOption.TopDirectoryOnly)) + { + File.Delete(path); + } + } + File.WriteAllText(CurrentRunPath, RunKey(runId), Encoding.ASCII); + WorkflowSpoolProtector.RestrictToCurrentUser(CurrentRunPath); + } + + public void CloseRun(string runId) + { + using var inboxLock = AcquireLock(); + if (IsCurrentRun(runId) && File.Exists(CurrentRunPath)) + File.Delete(CurrentRunPath); + } + + public async Task SubmitAsync( + DiagnosticSnapshotInboxRequest request, + TimeSpan acknowledgementTimeout, + CancellationToken cancellationToken) + { + var key = RequestKey(request.RunId, request.SnapshotId); + var requestPath = Path.Combine(_inboxDirectory, key + RequestSuffix); + var acknowledgementPath = Path.Combine(_inboxDirectory, key + AcknowledgementSuffix); + + await using (AcquireLock().ConfigureAwait(false)) + { + if (!IsCurrentRun(request.RunId)) + return Failure(request.SnapshotId, "run_closing"); + + var acknowledgement = ReadAcknowledgement(acknowledgementPath); + if (acknowledgement is not null) + return SubmissionFromAcknowledgement(request, acknowledgement); + + var pending = ReadRequest(requestPath); + if (pending is not null) + { + if (!string.Equals(pending.PayloadDigest, request.PayloadDigest, StringComparison.Ordinal)) + return Failure(request.SnapshotId, "snapshot_conflict"); + } + else + { + await WriteProtectedAtomicallyAsync( + requestPath, + JsonSerializer.SerializeToUtf8Bytes( + request, + CodexObserverStateJsonContext.Default.DiagnosticSnapshotInboxRequest), + overwrite: false, + cancellationToken).ConfigureAwait(false); + } + } + + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeout.CancelAfter(acknowledgementTimeout); + try + { + while (true) + { + var acknowledgement = ReadAcknowledgement(acknowledgementPath); + if (acknowledgement is not null) + return SubmissionFromAcknowledgement(request, acknowledgement); + await Task.Delay(s_pollInterval, timeout.Token).ConfigureAwait(false); + } + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + return Failure(request.SnapshotId, "ack_timeout"); + } + } + + public IReadOnlyList ReadPending(string runId) + { + var requests = new List(); + foreach (var path in Directory.EnumerateFiles( + _inboxDirectory, + $"*{RequestSuffix}", + SearchOption.TopDirectoryOnly)) + { + try + { + var request = ReadRequest(path); + if (request is not null && request.RunId == runId) + requests.Add(request); + } + catch (Exception exception) when (IsUnreadableEnvelope(exception)) + { + Quarantine(path); + Console.Error.WriteLine( + "[qyl] Ignored an unreadable diagnostic inbox request (diagnostic_inbox_unreadable)."); + } + } + return requests + .OrderBy(static request => request.SubmittedAt) + .ThenBy(static request => request.SnapshotId, StringComparer.Ordinal) + .ToArray(); + } + + public async Task AcknowledgeAsync( + DiagnosticSnapshotInboxRequest request, + string status, + string code, + string? eventId, + CancellationToken cancellationToken) + { + var acknowledgement = new DiagnosticSnapshotInboxAcknowledgement( + request.RunId, + request.SnapshotId, + request.PayloadDigest, + status, + code, + eventId); + var key = RequestKey(request.RunId, request.SnapshotId); + var acknowledgementPath = Path.Combine(_inboxDirectory, key + AcknowledgementSuffix); + await WriteProtectedAtomicallyAsync( + acknowledgementPath, + JsonSerializer.SerializeToUtf8Bytes( + acknowledgement, + CodexObserverStateJsonContext.Default.DiagnosticSnapshotInboxAcknowledgement), + overwrite: true, + cancellationToken).ConfigureAwait(false); + var requestPath = Path.Combine(_inboxDirectory, key + RequestSuffix); + if (File.Exists(requestPath)) + File.Delete(requestPath); + } + + private static DiagnosticSnapshotSubmissionResult SubmissionFromAcknowledgement( + DiagnosticSnapshotInboxRequest request, + DiagnosticSnapshotInboxAcknowledgement acknowledgement) + { + if (!string.Equals(acknowledgement.RunId, request.RunId, StringComparison.Ordinal) || + !string.Equals(acknowledgement.SnapshotId, request.SnapshotId, StringComparison.Ordinal) || + !string.Equals(acknowledgement.PayloadDigest, request.PayloadDigest, StringComparison.Ordinal)) + { + return Failure(request.SnapshotId, "snapshot_conflict"); + } + return new DiagnosticSnapshotSubmissionResult( + acknowledgement.Status == "recorded", + acknowledgement.Code, + request.SnapshotId, + acknowledgement.EventId); + } + + private DiagnosticSnapshotInboxRequest? ReadRequest(string path) + { + if (!File.Exists(path)) + return null; + var plaintext = ReadProtected(path); + try + { + return JsonSerializer.Deserialize( + plaintext, + CodexObserverStateJsonContext.Default.DiagnosticSnapshotInboxRequest) + ?? throw new InvalidDataException("The diagnostic inbox contains an empty request."); + } + finally + { + CryptographicOperations.ZeroMemory(plaintext); + } + } + + private DiagnosticSnapshotInboxAcknowledgement? ReadAcknowledgement(string path) + { + if (!File.Exists(path)) + return null; + var plaintext = ReadProtected(path); + try + { + return JsonSerializer.Deserialize( + plaintext, + CodexObserverStateJsonContext.Default.DiagnosticSnapshotInboxAcknowledgement) + ?? throw new InvalidDataException("The diagnostic inbox contains an empty acknowledgement."); + } + finally + { + CryptographicOperations.ZeroMemory(plaintext); + } + } + + private byte[] ReadProtected(string path) + { + var envelope = JsonSerializer.Deserialize( + File.ReadAllText(path), + CodexObserverStateJsonContext.Default.WorkflowSpoolEnvelope) + ?? throw new InvalidDataException("The diagnostic inbox contains an empty encrypted envelope."); + return _protector.Unprotect(envelope); + } + + private async Task WriteProtectedAtomicallyAsync( + string path, + byte[] plaintext, + bool overwrite, + CancellationToken cancellationToken) + { + var envelope = _protector.Protect(plaintext); + var encodedEnvelope = JsonSerializer.Serialize( + envelope, + CodexObserverStateJsonContext.Default.WorkflowSpoolEnvelope); + var temporaryPath = $"{path}.{Guid.NewGuid():N}.tmp"; + try + { + await File.WriteAllTextAsync( + temporaryPath, + encodedEnvelope, + Encoding.UTF8, + cancellationToken).ConfigureAwait(false); + WorkflowSpoolProtector.RestrictToCurrentUser(temporaryPath); + try + { + File.Move(temporaryPath, path, overwrite); + } + catch (IOException) when (!overwrite && File.Exists(path)) + { + return; + } + WorkflowSpoolProtector.RestrictToCurrentUser(path); + } + finally + { + CryptographicOperations.ZeroMemory(plaintext); + if (File.Exists(temporaryPath)) + File.Delete(temporaryPath); + } + } + + private string CurrentRunPath => Path.Combine(_inboxDirectory, CurrentRunFileName); + + private bool IsCurrentRun(string runId) => + File.Exists(CurrentRunPath) && + string.Equals(File.ReadAllText(CurrentRunPath), RunKey(runId), StringComparison.Ordinal); + + private FileStream AcquireLock() + { + var path = Path.Combine(_inboxDirectory, LockFileName); + while (true) + { + try + { + var stream = new FileStream( + path, + FileMode.OpenOrCreate, + FileAccess.ReadWrite, + FileShare.None, + 1, + FileOptions.WriteThrough); + WorkflowSpoolProtector.RestrictToCurrentUser(path); + return stream; + } + catch (IOException) + { + Thread.Sleep(10); + } + } + } + + private static string RequestKey(string runId, string snapshotId) => + Hash($"{runId}\n{snapshotId}"); + + private static string RunKey(string runId) => Hash(runId); + + private static string Hash(string value) => + Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(value))); + + private static DiagnosticSnapshotSubmissionResult Failure(string snapshotId, string code) => + new(false, code, snapshotId, null); + + private static bool IsUnreadableEnvelope(Exception exception) => + exception is IOException or InvalidDataException or CryptographicException or JsonException or + FormatException or ArgumentException or UnauthorizedAccessException; + + private static void Quarantine(string path) + { + try + { + File.Move(path, path + ".corrupt", overwrite: true); + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException) + { + // A later scan will ignore the same unreadable request again without aborting shutdown. + } + } + + private static void RestrictDirectory(string path) + { + if (!OperatingSystem.IsWindows()) + { + File.SetUnixFileMode( + path, + UnixFileMode.UserRead | UnixFileMode.UserWrite | UnixFileMode.UserExecute); + } + } +} diff --git a/packages/Qyl.Cli/Codex/ObserverBridgeServer.cs b/packages/Qyl.Cli/Codex/ObserverBridgeServer.cs index 4894b3ef..a76a9200 100644 --- a/packages/Qyl.Cli/Codex/ObserverBridgeServer.cs +++ b/packages/Qyl.Cli/Codex/ObserverBridgeServer.cs @@ -1,16 +1,19 @@ +using System.Security.Cryptography; +using System.Text; using System.Text.Json; namespace Qyl.Cli.Codex; internal static class ObserverBridgeServer { - private const int MaxMessageCharacters = 1024 * 1024; + private const int MaxMessageCharacters = 256 * 1024; private const string ProtocolVersion = "2026-07-28"; private const string ProtocolVersionMetaKey = "io.modelcontextprotocol/protocolVersion"; private const string ClientInfoMetaKey = "io.modelcontextprotocol/clientInfo"; private const string ClientCapabilitiesMetaKey = "io.modelcontextprotocol/clientCapabilities"; private const string ServerInfoMetaKey = "io.modelcontextprotocol/serverInfo"; - private const string ToolName = "get_active_workflow_run"; + private const string ReadToolName = "get_active_workflow_run"; + private const string DiagnosticToolName = "record_diagnostic_snapshot"; public static async Task RunAsync( ActiveWorkflowRunStore activeRuns, @@ -18,21 +21,30 @@ public static async Task RunAsync( TextWriter output, CancellationToken cancellationToken) { + var diagnosticInbox = new DiagnosticSnapshotInbox(activeRuns.Root); + var requestReader = new BoundedLineReader(input, MaxMessageCharacters); while (!cancellationToken.IsCancellationRequested) { - var line = await input.ReadLineAsync(cancellationToken).ConfigureAwait(false); - if (line is null) + var requestLine = await requestReader.ReadAsync(cancellationToken).ConfigureAwait(false); + if (requestLine.EndOfStream) return 0; - if (line.Length > MaxMessageCharacters) + if (requestLine.ExceedsLimit) { - await WriteErrorAsync(output, null, -32600, "MCP request exceeds the 1 MiB limit.") + await WriteErrorAsync( + output, + null, + -32600, + "MCP request exceeds the 262,144-character transport limit.") .ConfigureAwait(false); continue; } + var line = requestLine.Value!; try { - using var document = JsonDocument.Parse(line); + using var document = JsonDocument.Parse( + line, + new JsonDocumentOptions { MaxDepth = 16 }); var request = document.RootElement; if (!request.TryGetProperty("method", out var methodElement) || methodElement.ValueKind is not JsonValueKind.String) @@ -90,7 +102,13 @@ await WriteInvalidEnvelopeAsync( await WriteToolsAsync(output, id).ConfigureAwait(false); break; case "tools/call": - await WriteToolResultAsync(output, id, request, activeRuns.Read()) + await WriteToolResultAsync( + output, + id, + request, + activeRuns, + diagnosticInbox, + cancellationToken) .ConfigureAwait(false); break; default: @@ -132,7 +150,7 @@ await WriteAsync( writer.WriteEndObject(); writer.WriteString( "instructions", - "Read the active qyl Codex workflow run through get_active_workflow_run."); + "Read the active qyl Codex workflow run or record a bounded diagnostic snapshot against it."); WritePublicCache(writer); }).ConfigureAwait(false); } @@ -147,7 +165,7 @@ await WriteAsync( writer.WritePropertyName("tools"); writer.WriteStartArray(); writer.WriteStartObject(); - writer.WriteString("name", ToolName); + writer.WriteString("name", ReadToolName); writer.WriteString( "description", "Returns the workflow run observed by the active qyl codex process, if one exists."); @@ -190,6 +208,7 @@ await WriteAsync( writer.WriteBoolean("openWorldHint", false); writer.WriteEndObject(); writer.WriteEndObject(); + WriteDiagnosticToolSchema(writer); writer.WriteEndArray(); WritePublicCache(writer); }).ConfigureAwait(false); @@ -199,17 +218,41 @@ private static async Task WriteToolResultAsync( TextWriter output, JsonElement? id, JsonElement request, - ActiveWorkflowRun? active) + ActiveWorkflowRunStore activeRuns, + DiagnosticSnapshotInbox diagnosticInbox, + CancellationToken cancellationToken) { if (!request.TryGetProperty("params", out var parameters) || !parameters.TryGetProperty("name", out var nameElement) || - nameElement.GetString() != ToolName) + nameElement.ValueKind is not JsonValueKind.String) { - await WriteErrorAsync(output, id, -32602, $"Unknown tool. Expected '{ToolName}'.") + await WriteErrorAsync(output, id, -32602, "Tool name is required.") .ConfigureAwait(false); return; } + var toolName = nameElement.GetString(); + if (toolName == DiagnosticToolName) + { + await WriteDiagnosticToolResultAsync( + output, + id, + parameters, + activeRuns.Read(), + diagnosticInbox, + cancellationToken) + .ConfigureAwait(false); + return; + } + if (toolName != ReadToolName) + { + await WriteErrorAsync(output, id, -32602, $"Unknown tool '{toolName}'.") + .ConfigureAwait(false); + return; + } + + var active = activeRuns.Read(); + await WriteAsync( output, id, @@ -242,12 +285,291 @@ active is null }).ConfigureAwait(false); } + private static async Task WriteDiagnosticToolResultAsync( + TextWriter output, + JsonElement? id, + JsonElement parameters, + ActiveWorkflowRun? active, + DiagnosticSnapshotInbox diagnosticInbox, + CancellationToken cancellationToken) + { + if (!parameters.TryGetProperty("arguments", out var arguments) || + arguments.ValueKind is not JsonValueKind.Object) + { + await WriteDiagnosticResultAsync( + output, + id, + new DiagnosticSnapshotSubmissionResult(false, "invalid_input", "unknown", null), + "arguments") + .ConfigureAwait(false); + return; + } + + var snapshotId = arguments.TryGetProperty("snapshotId", out var snapshotElement) && + snapshotElement.ValueKind is JsonValueKind.String + ? snapshotElement.GetString() ?? "unknown" + : "unknown"; + if (active is null) + { + await WriteDiagnosticResultAsync( + output, + id, + new DiagnosticSnapshotSubmissionResult(false, "no_active_run", snapshotId, null), + null) + .ConfigureAwait(false); + return; + } + if (active.ThreadId is null) + { + await WriteDiagnosticResultAsync( + output, + id, + new DiagnosticSnapshotSubmissionResult( + false, + "context_unavailable", + snapshotId, + null), + null) + .ConfigureAwait(false); + return; + } + + DiagnosticSnapshotInboxRequest? request; + DiagnosticSnapshotValidationError validationError; + try + { + if (!DiagnosticSnapshotCapture.TryCreate( + active, + arguments, + diagnosticInbox.Protector, + TimeProvider.System.GetUtcNow(), + out request, + out validationError)) + { + await WriteDiagnosticResultAsync( + output, + id, + new DiagnosticSnapshotSubmissionResult( + false, + validationError.Code, + snapshotId, + null), + validationError.Field) + .ConfigureAwait(false); + return; + } + } + catch (InvalidDataException) + { + await WriteDiagnosticResultAsync( + output, + id, + new DiagnosticSnapshotSubmissionResult( + false, + "invalid_json_value", + snapshotId, + null), + "variables") + .ConfigureAwait(false); + return; + } + + DiagnosticSnapshotSubmissionResult result; + try + { + result = await diagnosticInbox.SubmitAsync( + request!, + TimeSpan.FromSeconds(3), + cancellationToken) + .ConfigureAwait(false); + } + catch (Exception exception) when ( + exception is IOException or InvalidDataException or CryptographicException) + { + result = new DiagnosticSnapshotSubmissionResult( + false, + "inbox_failure", + request!.SnapshotId, + null); + } + await WriteDiagnosticResultAsync(output, id, result, null).ConfigureAwait(false); + } + + private static Task WriteDiagnosticResultAsync( + TextWriter output, + JsonElement? id, + DiagnosticSnapshotSubmissionResult result, + string? field) => + WriteAsync( + output, + id, + writer => + { + writer.WritePropertyName("content"); + writer.WriteStartArray(); + writer.WriteStartObject(); + writer.WriteString("type", "text"); + writer.WriteString("text", result.Code); + writer.WriteEndObject(); + writer.WriteEndArray(); + writer.WritePropertyName("structuredContent"); + writer.WriteStartObject(); + writer.WriteBoolean("recorded", result.Recorded); + writer.WriteString("code", result.Code); + writer.WriteString("snapshotId", result.SnapshotId); + if (result.EventId is not null) + writer.WriteString("eventId", result.EventId); + if (field is not null) + writer.WriteString("field", field); + writer.WriteEndObject(); + writer.WriteBoolean("isError", !result.Recorded); + }); + private static Task WriteEmptyResultAsync(TextWriter output, JsonElement? id) => WriteAsync( output, id, static _ => { }); + private static void WriteDiagnosticToolSchema(Utf8JsonWriter writer) + { + writer.WriteStartObject(); + writer.WriteString("name", DiagnosticToolName); + writer.WriteString( + "description", + "Record one immutable, bounded diagnostic state frame against the active qyl Codex run. " + + "Reuse the same snapshotId and payload when retrying; changing a payload under an existing " + + "snapshotId is a conflict. Variable names remain data: public/internal values enter protected " + + "content, sensitive values are redacted, and secret values are omitted. Checks reference " + + "variable names and use closed operators; expression strings are not accepted."); + writer.WritePropertyName("inputSchema"); + writer.WriteStartObject(); + writer.WriteString("$schema", "https://json-schema.org/draft/2020-12/schema"); + writer.WriteString("type", "object"); + writer.WritePropertyName("properties"); + writer.WriteStartObject(); + WriteMachineIdentifierSchema(writer, "snapshotId"); + WriteMachineIdentifierSchema(writer, "probeId"); + WriteEnumSchema(writer, "phase", ["input", "output", "error", "checkpoint"]); + writer.WritePropertyName("variables"); + writer.WriteStartObject(); + writer.WriteString("type", "array"); + writer.WriteString( + "description", + "Dynamically named typed state. Classification is applied before the frame leaves the bridge process."); + writer.WriteNumber("maxItems", DiagnosticSnapshotCapture.MaxVariables); + writer.WritePropertyName("items"); + writer.WriteStartObject(); + writer.WriteString("type", "object"); + writer.WritePropertyName("properties"); + writer.WriteStartObject(); + WriteMachineIdentifierSchema(writer, "name"); + WriteEnumSchema(writer, "classification", ["public", "internal", "sensitive", "secret"]); + writer.WritePropertyName("value"); + writer.WriteStartObject(); + writer.WriteEndObject(); + writer.WriteEndObject(); + WriteRequired(writer, ["name", "classification", "value"]); + writer.WriteBoolean("additionalProperties", false); + writer.WriteEndObject(); + writer.WriteEndObject(); + writer.WritePropertyName("checks"); + writer.WriteStartObject(); + writer.WriteString("type", "array"); + writer.WriteString( + "description", + "Structural checks whose actual/expected operands are variable-name references. Missing or incompatible operands are unknown, except exists on absent/null state is fail."); + writer.WriteNumber("maxItems", DiagnosticSnapshotCapture.MaxChecks); + writer.WritePropertyName("default"); + writer.WriteStartArray(); + writer.WriteEndArray(); + writer.WritePropertyName("items"); + writer.WriteStartObject(); + writer.WriteString("type", "object"); + writer.WritePropertyName("properties"); + writer.WriteStartObject(); + WriteMachineIdentifierSchema(writer, "checkId"); + WriteEnumSchema( + writer, + "operator", + ["equal", "not_equal", "exists", "type_is", "contains", "less_than", "greater_than"]); + WriteMachineIdentifierSchema(writer, "actual"); + WriteMachineIdentifierSchema(writer, "expected"); + WriteEnumSchema( + writer, + "expectedType", + ["null", "boolean", "integer", "number", "string", "json"]); + writer.WriteEndObject(); + WriteRequired(writer, ["checkId", "operator", "actual"]); + writer.WriteBoolean("additionalProperties", false); + writer.WriteEndObject(); + writer.WriteEndObject(); + writer.WriteEndObject(); + WriteRequired(writer, ["snapshotId", "probeId", "phase", "variables"]); + writer.WriteBoolean("additionalProperties", false); + writer.WriteEndObject(); + + writer.WritePropertyName("outputSchema"); + writer.WriteStartObject(); + writer.WriteString("$schema", "https://json-schema.org/draft/2020-12/schema"); + writer.WriteString("type", "object"); + writer.WritePropertyName("properties"); + writer.WriteStartObject(); + WriteBooleanSchema(writer, "recorded"); + WriteStringSchema(writer, "code"); + WriteStringSchema(writer, "snapshotId"); + WriteStringSchema(writer, "eventId"); + WriteStringSchema(writer, "field"); + writer.WriteEndObject(); + WriteRequired(writer, ["recorded", "code", "snapshotId"]); + writer.WriteBoolean("additionalProperties", false); + writer.WriteEndObject(); + writer.WritePropertyName("annotations"); + writer.WriteStartObject(); + writer.WriteBoolean("readOnlyHint", false); + writer.WriteBoolean("destructiveHint", false); + writer.WriteBoolean("idempotentHint", true); + writer.WriteBoolean("openWorldHint", false); + writer.WriteEndObject(); + writer.WriteEndObject(); + } + + private static void WriteMachineIdentifierSchema(Utf8JsonWriter writer, string name) + { + writer.WritePropertyName(name); + writer.WriteStartObject(); + writer.WriteString("type", "string"); + writer.WriteNumber("minLength", 1); + writer.WriteNumber("maxLength", DiagnosticSnapshotCapture.MaxIdentifierLength); + writer.WriteString("pattern", "^[A-Za-z0-9_][A-Za-z0-9._:/\\[\\]-]*$"); + writer.WriteEndObject(); + } + + private static void WriteEnumSchema( + Utf8JsonWriter writer, + string name, + IReadOnlyList values) + { + writer.WritePropertyName(name); + writer.WriteStartObject(); + writer.WriteString("type", "string"); + writer.WritePropertyName("enum"); + writer.WriteStartArray(); + foreach (var value in values) + writer.WriteStringValue(value); + writer.WriteEndArray(); + writer.WriteEndObject(); + } + + private static void WriteRequired(Utf8JsonWriter writer, IReadOnlyList names) + { + writer.WritePropertyName("required"); + writer.WriteStartArray(); + foreach (var name in names) + writer.WriteStringValue(name); + writer.WriteEndArray(); + } + private static bool TryValidateModernEnvelope( JsonElement request, out string invalidKey, @@ -434,6 +756,72 @@ private static void WriteBooleanSchema(Utf8JsonWriter writer, string name) writer.WriteEndObject(); } + private readonly record struct BoundedLine(string? Value, bool ExceedsLimit) + { + public bool EndOfStream => Value is null && !ExceedsLimit; + } + + private sealed class BoundedLineReader(TextReader reader, int maximumCharacters) + { + private readonly char[] _buffer = new char[4 * 1024]; + private int _count; + private int _offset; + + public async ValueTask ReadAsync(CancellationToken cancellationToken) + { + var value = new StringBuilder(Math.Min(maximumCharacters, _buffer.Length)); + var sawCharacters = false; + var exceedsLimit = false; + while (true) + { + if (_offset == _count) + { + _count = await reader.ReadAsync(_buffer.AsMemory(), cancellationToken) + .ConfigureAwait(false); + _offset = 0; + if (_count == 0) + { + return !sawCharacters + ? new BoundedLine(null, false) + : Complete(value, exceedsLimit); + } + } + + var remaining = _buffer.AsSpan(_offset, _count - _offset); + var newline = remaining.IndexOf('\n'); + var segmentLength = newline < 0 ? remaining.Length : newline; + sawCharacters |= segmentLength > 0; + if (!exceedsLimit) + { + if (segmentLength > maximumCharacters - value.Length) + { + exceedsLimit = true; + value.Clear(); + } + else + { + value.Append(remaining[..segmentLength]); + } + } + _offset += segmentLength; + + if (newline < 0) + continue; + _offset++; + return Complete(value, exceedsLimit); + } + } + + private static BoundedLine Complete(StringBuilder value, bool exceedsLimit) + { + if (exceedsLimit) + return new BoundedLine(null, true); + if (value.Length > 0 && value[^1] == '\r') + value.Length--; + return new BoundedLine(value.ToString(), false); + } + } + private static void WriteStringSchema(Utf8JsonWriter writer, string name) { writer.WritePropertyName(name); diff --git a/packages/Qyl.Cli/Codex/WorkflowJournalPump.cs b/packages/Qyl.Cli/Codex/WorkflowJournalPump.cs index 083d9257..39ecb497 100644 --- a/packages/Qyl.Cli/Codex/WorkflowJournalPump.cs +++ b/packages/Qyl.Cli/Codex/WorkflowJournalPump.cs @@ -76,6 +76,7 @@ public async Task RunControlLoopAsync( string runId, CodexEventNormalizer normalizer, ICodexControlClient appServer, + SemaphoreSlim? normalizerGate, CancellationToken cancellationToken) { ulong cursor = 0; @@ -126,9 +127,26 @@ await collector.UpdateControlAsync( if (!_appliedControls.TryGetValue(command.CommandId, out var appliedAt)) { + CodexControlTarget target; + if (normalizerGate is null) + { + target = normalizer.ControlTarget; + } + else + { + await normalizerGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + target = normalizer.ControlTarget; + } + finally + { + normalizerGate.Release(); + } + } await ApplyControlAsync( command, - normalizer.ControlTarget, + target, appServer, cancellationToken).ConfigureAwait(false); appliedAt = TimeProvider.System.GetUtcNow(); diff --git a/packages/Qyl.Cli/Codex/WorkflowSpool.cs b/packages/Qyl.Cli/Codex/WorkflowSpool.cs index 0e9e8521..c97f45ba 100644 --- a/packages/Qyl.Cli/Codex/WorkflowSpool.cs +++ b/packages/Qyl.Cli/Codex/WorkflowSpool.cs @@ -62,23 +62,36 @@ public async Task AppendAsync(WorkflowSpoolEntry entry, CancellationToken cancel var line = JsonSerializer.SerializeToUtf8Bytes( envelope, CodexObserverStateJsonContext.Default.WorkflowSpoolEnvelope); + var record = GC.AllocateUninitializedArray(line.Length + 1); + line.CopyTo(record, 0); + record[^1] = (byte)'\n'; await _writeLock.Reader.ReadAsync(cancellationToken).ConfigureAwait(false); try { var stream = new FileStream( EventsPath, - FileMode.Append, - FileAccess.Write, + FileMode.OpenOrCreate, + FileAccess.ReadWrite, FileShare.Read, 16 * 1024, FileOptions.Asynchronous | FileOptions.WriteThrough); await using (stream.ConfigureAwait(false)) { - await stream.WriteAsync(line, cancellationToken).ConfigureAwait(false); - await stream.WriteAsync("\n"u8.ToArray(), cancellationToken).ConfigureAwait(false); - await stream.FlushAsync(cancellationToken).ConfigureAwait(false); WorkflowSpoolProtector.RestrictToCurrentUser(EventsPath); + var originalLength = stream.Length; + stream.Position = originalLength; + try + { + await stream.WriteAsync(record, cancellationToken).ConfigureAwait(false); + await stream.FlushAsync(cancellationToken).ConfigureAwait(false); + } + catch + { + stream.SetLength(originalLength); + stream.Flush(flushToDisk: true); + throw; + } } } finally diff --git a/packages/Qyl.Cli/Codex/WorkflowSpoolProtector.cs b/packages/Qyl.Cli/Codex/WorkflowSpoolProtector.cs index 763d7668..6e42034f 100644 --- a/packages/Qyl.Cli/Codex/WorkflowSpoolProtector.cs +++ b/packages/Qyl.Cli/Codex/WorkflowSpoolProtector.cs @@ -81,6 +81,12 @@ public byte[] Unprotect(WorkflowSpoolEnvelope envelope) return plaintext; } + public string KeyedDigest(ReadOnlySpan value) + { + var digest = HMACSHA256.HashData(_key, value); + return $"hmac-sha256:{Convert.ToHexStringLower(digest)}"; + } + internal static void RestrictToCurrentUser(string path) { if (OperatingSystem.IsWindows()) diff --git a/packages/Qyl.Cli/Codex/WorkflowTelemetryProjection.cs b/packages/Qyl.Cli/Codex/WorkflowTelemetryProjection.cs index 3954c022..9ced96cf 100644 --- a/packages/Qyl.Cli/Codex/WorkflowTelemetryProjection.cs +++ b/packages/Qyl.Cli/Codex/WorkflowTelemetryProjection.cs @@ -13,6 +13,21 @@ namespace Qyl.Cli.Codex; internal sealed class WorkflowTelemetryProjection : IDisposable { private const string SourceName = "qyl.codex.observer"; + private const string DiagnosticEventName = "qyl.agent.diagnostic.snapshot"; + private const string DiagnosticExtensionId = "qyl.agent.diagnostic.extension.id"; + private const string DiagnosticFormatVersion = "qyl.agent.diagnostic.format.version"; + private const string DiagnosticSnapshotId = "qyl.agent.diagnostic.snapshot.id"; + private const string DiagnosticProbeId = "qyl.agent.diagnostic.probe.id"; + private const string DiagnosticPhase = "qyl.agent.diagnostic.phase"; + private const string DiagnosticOutcome = "qyl.agent.diagnostic.outcome"; + private const string DiagnosticVariableCount = "qyl.agent.diagnostic.variable.count"; + private const string DiagnosticCheckCount = "qyl.agent.diagnostic.check.count"; + private const string DiagnosticFailedCheckCount = "qyl.agent.diagnostic.check.failed_count"; + private const string WorkflowRunId = "qyl.workflow.run.id"; + private const string WorkflowEventId = "qyl.workflow.event.id"; + private const string WorkflowAttemptId = "qyl.workflow.attempt.id"; + private const string WorkflowAgentId = "qyl.workflow.agent.id"; + private const string WorkflowToolCallId = "qyl.workflow.tool_call.id"; private static readonly Action s_logJournalEvent = LoggerMessage.Define( LogLevel.Information, @@ -20,13 +35,15 @@ internal sealed class WorkflowTelemetryProjection : IDisposable "Workflow journal event {EventKind} {EventId}"); private readonly ActivitySource _source = new(SourceName, BuildVersion.ProductVersion); + private readonly string _runId; private readonly Dictionary _activities = new(StringComparer.Ordinal); private readonly TracerProvider? _traces; private readonly ILoggerFactory? _loggerFactory; private readonly ILogger? _logger; - private WorkflowTelemetryProjection(string? apiKey) + private WorkflowTelemetryProjection(string runId, string? apiKey) { + _runId = runId; if (string.IsNullOrWhiteSpace(apiKey)) return; @@ -62,7 +79,8 @@ private WorkflowTelemetryProjection(string? apiKey) _logger = _loggerFactory.CreateLogger(SourceName); } - public static WorkflowTelemetryProjection Create(string? apiKey) => new(apiKey); + public static WorkflowTelemetryProjection Create(string runId, string? apiKey) => + new(runId, apiKey); public void Record(WorkflowEventAppend workflowEvent) { @@ -113,6 +131,9 @@ public void Record(WorkflowEventAppend workflowEvent) case WorkflowJournalEventKind.Joined: RecordJoin(workflowEvent); break; + case WorkflowJournalEventKind.ContentCaptured: + RecordDiagnosticSnapshot(workflowEvent); + break; case WorkflowJournalEventKind.ToolCompleted: Stop(ToolKey(workflowEvent), workflowEvent); break; @@ -199,6 +220,56 @@ private void RecordJoin(WorkflowEventAppend workflowEvent) activity?.SetEndTime(workflowEvent.Timestamp.UtcDateTime); } + private void RecordDiagnosticSnapshot(WorkflowEventAppend workflowEvent) + { + if (DataString(workflowEvent, "extension_id") != DiagnosticSnapshotCapture.ExtensionId || + DataInt64(workflowEvent, "format_version") != DiagnosticSnapshotCapture.FormatVersion) + { + return; + } + var activity = ActiveActivity(workflowEvent); + if (activity is null) + return; + + var tags = new ActivityTagsCollection + { + [DiagnosticExtensionId] = DataString(workflowEvent, "extension_id"), + [DiagnosticFormatVersion] = DataInt64(workflowEvent, "format_version"), + [DiagnosticSnapshotId] = DataString(workflowEvent, "snapshot_id"), + [DiagnosticProbeId] = DataString(workflowEvent, "probe_id"), + [DiagnosticPhase] = DataString(workflowEvent, "phase"), + [DiagnosticOutcome] = DataString(workflowEvent, "outcome"), + [DiagnosticVariableCount] = DataInt64(workflowEvent, "variable_count"), + [DiagnosticCheckCount] = DataInt64(workflowEvent, "check_count"), + [DiagnosticFailedCheckCount] = DataInt64(workflowEvent, "failed_check_count"), + [WorkflowRunId] = _runId, + [WorkflowEventId] = workflowEvent.EventId.Value + }; + if (workflowEvent.AttemptId is not null) + tags[WorkflowAttemptId] = workflowEvent.AttemptId.Value.Value; + if (workflowEvent.AgentId is not null) + tags[WorkflowAgentId] = workflowEvent.AgentId.Value.Value; + if (workflowEvent.ToolCallId is not null) + tags[WorkflowToolCallId] = workflowEvent.ToolCallId.Value.Value; + activity.AddEvent(new ActivityEvent(DiagnosticEventName, workflowEvent.Timestamp, tags)); + } + + private Activity? ActiveActivity(WorkflowEventAppend workflowEvent) + { + foreach (var key in new[] + { + AgentKey(workflowEvent.AgentId), + TurnKey(workflowEvent), + AttemptKey(workflowEvent), + "run" + }) + { + if (key is not null && _activities.TryGetValue(key, out var activity)) + return activity; + } + return null; + } + private ActivityContext AgentParentContext(WorkflowEventAppend workflowEvent) { var parent = Context(AgentKey(workflowEvent.ParentAgentId)); @@ -253,6 +324,34 @@ private ActivityContext Context(string? key) => : value.ToString(); } + private static string? DataString(WorkflowEventAppend workflowEvent, string key) + { + if (workflowEvent.Data is null || !workflowEvent.Data.TryGetValue(key, out var value)) + return null; + return value is JsonElement { ValueKind: JsonValueKind.String } element + ? element.GetString() + : value?.ToString(); + } + + private static long? DataInt64(WorkflowEventAppend workflowEvent, string key) + { + if (workflowEvent.Data is null || !workflowEvent.Data.TryGetValue(key, out var value)) + return null; + if (value is JsonElement { ValueKind: JsonValueKind.Number } element && + element.TryGetInt64(out var jsonValue)) + { + return jsonValue; + } + return value switch + { + byte item => item, + short item => item, + int item => item, + long item => item, + _ => null + }; + } + private static string? AttemptKey(WorkflowEventAppend workflowEvent) => workflowEvent.AttemptId is null ? null : $"attempt:{workflowEvent.AttemptId.Value.Value}"; diff --git a/services/qyl.collector/Ingestion/Generated/CollectorSemanticAttributeCatalog.g.cs b/services/qyl.collector/Ingestion/Generated/CollectorSemanticAttributeCatalog.g.cs index 8386417e..08008eff 100644 --- a/services/qyl.collector/Ingestion/Generated/CollectorSemanticAttributeCatalog.g.cs +++ b/services/qyl.collector/Ingestion/Generated/CollectorSemanticAttributeCatalog.g.cs @@ -248,6 +248,15 @@ internal static class CollectorSemanticAttributeCatalog "otel.scope.version", "otel.status_code", "otel.status_description", + "qyl.agent.diagnostic.check.count", // incubating + "qyl.agent.diagnostic.check.failed_count", // incubating + "qyl.agent.diagnostic.extension.id", // incubating + "qyl.agent.diagnostic.format.version", // incubating + "qyl.agent.diagnostic.outcome", // incubating + "qyl.agent.diagnostic.phase", // incubating + "qyl.agent.diagnostic.probe.id", // incubating + "qyl.agent.diagnostic.snapshot.id", // incubating + "qyl.agent.diagnostic.variable.count", // incubating "qyl.exception.source", // incubating "qyl.instrumentation.domain", // incubating "qyl.mcp.evaluation_run.id", // incubating @@ -258,6 +267,11 @@ internal static class CollectorSemanticAttributeCatalog "qyl.mcp.server.id", // incubating "qyl.mcp.test_case.id", // incubating "qyl.mcp.tool.name", // incubating + "qyl.workflow.agent.id", // incubating + "qyl.workflow.attempt.id", // incubating + "qyl.workflow.event.id", // incubating + "qyl.workflow.run.id", // incubating + "qyl.workflow.tool_call.id", // incubating "rpc.connect_rpc.error_code", "rpc.connect_rpc.request.metadata", "rpc.connect_rpc.response.metadata", diff --git a/tests/Qyl.Cli.Tests/DiagnosticSnapshotTests.cs b/tests/Qyl.Cli.Tests/DiagnosticSnapshotTests.cs new file mode 100644 index 00000000..c2af898a --- /dev/null +++ b/tests/Qyl.Cli.Tests/DiagnosticSnapshotTests.cs @@ -0,0 +1,786 @@ +using System.Globalization; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using Qyl.Api.Contracts.Workflow; +using Qyl.Cli.Codex; + +namespace Qyl.Cli.Tests; + +public sealed class DiagnosticSnapshotTests +{ + private static readonly DateTimeOffset s_timestamp = + new(2026, 8, 9, 12, 0, 0, TimeSpan.Zero); + + [Fact] + public async Task Bridge_discovers_the_bounded_diagnostic_tool() + { + var root = TemporaryDirectory(); + try + { + var store = new ActiveWorkflowRunStore(root); + using var input = new StringReader( + """ + {"jsonrpc":"2.0","id":1,"method":"tools/list","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}} + """); + using var output = new StringWriter(CultureInfo.InvariantCulture); + + Assert.Equal( + 0, + await ObserverBridgeServer.RunAsync( + store, + input, + output, + TestContext.Current.CancellationToken)); + + using var response = JsonDocument.Parse(output.ToString()); + var tool = response.RootElement + .GetProperty("result") + .GetProperty("tools") + .EnumerateArray() + .Single(static item => item.GetProperty("name").GetString() == "record_diagnostic_snapshot"); + var schema = tool.GetProperty("inputSchema"); + var description = tool.GetProperty("description").GetString(); + Assert.Contains("Reuse the same snapshotId", description, StringComparison.Ordinal); + Assert.Contains("sensitive values are redacted", description, StringComparison.Ordinal); + Assert.Contains("secret values are omitted", description, StringComparison.Ordinal); + Assert.False(schema.GetProperty("additionalProperties").GetBoolean()); + Assert.Equal(64, schema.GetProperty("properties").GetProperty("variables").GetProperty("maxItems").GetInt32()); + Assert.Equal(64, schema.GetProperty("properties").GetProperty("checks").GetProperty("maxItems").GetInt32()); + Assert.False(tool.GetProperty("annotations").GetProperty("readOnlyHint").GetBoolean()); + Assert.True(tool.GetProperty("annotations").GetProperty("idempotentHint").GetBoolean()); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public async Task Bridge_discards_oversized_lines_and_resynchronizes_at_the_next_request() + { + var root = TemporaryDirectory(); + try + { + var oversized = new string('x', 256 * 1024 + 1); + using var input = new StringReader( + oversized + + "\n" + + "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"ping\",\"params\":{\"_meta\":{\"io.modelcontextprotocol/protocolVersion\":\"2026-07-28\",\"io.modelcontextprotocol/clientCapabilities\":{}}}}\n"); + using var output = new StringWriter(CultureInfo.InvariantCulture); + + Assert.Equal( + 0, + await ObserverBridgeServer.RunAsync( + new ActiveWorkflowRunStore(root), + input, + output, + TestContext.Current.CancellationToken)); + + var responses = output.ToString().Split( + Environment.NewLine, + StringSplitOptions.RemoveEmptyEntries); + Assert.Equal(2, responses.Length); + using var rejected = JsonDocument.Parse(responses[0]); + Assert.Equal(-32600, rejected.RootElement.GetProperty("error").GetProperty("code").GetInt32()); + using var pong = JsonDocument.Parse(responses[1]); + Assert.Equal(2, pong.RootElement.GetProperty("id").GetInt32()); + Assert.Equal("complete", pong.RootElement.GetProperty("result").GetProperty("resultType").GetString()); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public void Capture_is_typed_deterministic_and_redacts_before_handoff() + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + using var document = JsonDocument.Parse(ValidArgumentsJson); + Assert.True( + DiagnosticSnapshotCapture.TryCreate( + ActiveRun(), + document.RootElement, + inbox.Protector, + s_timestamp, + out var request, + out var error), + error.Code); + + Assert.NotNull(request); + Assert.Equal("fail", request.Outcome); + Assert.Equal(6, request.VariableCount); + Assert.Equal(4, request.CheckCount); + Assert.Equal(1, request.FailedCheckCount); + Assert.DoesNotContain("sensitive-plain", request.Content.Content, StringComparison.Ordinal); + Assert.DoesNotContain("secret-plain", request.Content.Content, StringComparison.Ordinal); + + using var captured = JsonDocument.Parse(request.Content.Content); + var payload = captured.RootElement; + Assert.Equal("qyl.agent.diagnostic.snapshot", payload.GetProperty("extension_id").GetString()); + Assert.Equal(1, payload.GetProperty("format_version").GetInt32()); + Assert.Matches("^[0-9a-f]{32}$", payload.GetProperty("capture_nonce").GetString()!); + Assert.Equal( + ["label", "limit", "missing", "password", "result", "token"], + payload.GetProperty("variables") + .EnumerateArray() + .Select(static item => item.GetProperty("name").GetString())); + + var sensitive = payload.GetProperty("variables").EnumerateArray() + .Single(static item => item.GetProperty("name").GetString() == "token"); + Assert.Equal("redacted", sensitive.GetProperty("capture").GetString()); + Assert.False(sensitive.TryGetProperty("value", out _)); + var secret = payload.GetProperty("variables").EnumerateArray() + .Single(static item => item.GetProperty("name").GetString() == "password"); + Assert.Equal("omitted", secret.GetProperty("capture").GetString()); + Assert.False(secret.TryGetProperty("value", out _)); + var result = payload.GetProperty("variables").EnumerateArray() + .Single(static item => item.GetProperty("name").GetString() == "result"); + Assert.Equal("integer", result.GetProperty("type").GetString()); + Assert.Equal("value", result.GetProperty("capture").GetString()); + + var contentHash = Convert.ToHexStringLower( + SHA256.HashData(Encoding.UTF8.GetBytes(request.Content.Content))); + Assert.Equal($"sha256:{contentHash}", request.Content.ContentRef.Value); + + var normalizer = SeedNormalizer(); + var batch = normalizer.NormalizeDiagnosticSnapshot(request, s_timestamp); + var workflowEvent = Assert.Single(batch.Events); + Assert.Equal(WorkflowJournalEventKind.ContentCaptured, workflowEvent.Kind); + Assert.Equal("thread-root", workflowEvent.ThreadId); + Assert.Equal("turn-root", workflowEvent.TurnId); + Assert.Equal("turn-root", workflowEvent.AttemptId?.Value); + Assert.Equal( + [ + "check_count", + "content_ref", + "extension_id", + "failed_check_count", + "format_version", + "outcome", + "phase", + "probe_id", + "snapshot_id", + "variable_count" + ], + workflowEvent.Data!.Keys.Order(StringComparer.Ordinal)); + Assert.Equal(request.Content.ContentRef.Value, workflowEvent.Data["content_ref"]); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public async Task Tool_call_crosses_encrypted_inbox_and_reaches_encrypted_spool() + { + var root = TemporaryDirectory(); + try + { + var activeRuns = new ActiveWorkflowRunStore(root); + await activeRuns.WriteAsync( + ActiveRun(), + TestContext.Current.CancellationToken); + var inbox = new DiagnosticSnapshotInbox(root); + inbox.PrepareRun("run-live"); + var spool = new WorkflowSpoolStore(root).Open("run-live"); + var normalizer = SeedNormalizer(); + using var journalGate = new SemaphoreSlim(1, 1); + using var drainCancellation = CancellationTokenSource.CreateLinkedTokenSource( + TestContext.Current.CancellationToken); + var drain = Task.Run( + async () => + { + try + { + while (!drainCancellation.IsCancellationRequested) + { + await CodexObserverRuntime.DrainDiagnosticsOnceAsync( + inbox, + "run-live", + normalizer, + spool, + null, + journalGate, + drainCancellation.Token); + await Task.Delay(10, drainCancellation.Token); + } + } + catch (OperationCanceledException) + { + } + }, + drainCancellation.Token); + + var compactArguments = string.Concat( + ValidArgumentsJson.Where(static character => character is not '\r' and not '\n')); + var requestJson = + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"record_diagnostic_snapshot\",\"arguments\":" + + compactArguments + + ",\"_meta\":{\"io.modelcontextprotocol/protocolVersion\":\"2026-07-28\",\"io.modelcontextprotocol/clientCapabilities\":{}}}}\n"; + using var input = new StringReader(requestJson); + using var output = new StringWriter(CultureInfo.InvariantCulture); + Assert.Equal( + 0, + await ObserverBridgeServer.RunAsync( + activeRuns, + input, + output, + TestContext.Current.CancellationToken)); + await drainCancellation.CancelAsync(); + await drain; + + using var response = JsonDocument.Parse(output.ToString()); + var result = response.RootElement.GetProperty("result"); + Assert.False(result.GetProperty("isError").GetBoolean()); + Assert.True(result.GetProperty("structuredContent").GetProperty("recorded").GetBoolean()); + Assert.Equal("recorded", result.GetProperty("structuredContent").GetProperty("code").GetString()); + + var entry = Assert.Single(spool.ReadAfter(0, 10)); + Assert.Equal(WorkflowJournalEventKind.ContentCaptured, entry.Event.Kind); + Assert.Single(entry.Content); + Assert.Contains("public-result", entry.Content[0].Content, StringComparison.Ordinal); + Assert.Contains("sensitive-plain", ValidArgumentsJson, StringComparison.Ordinal); + Assert.DoesNotContain("sensitive-plain", entry.Content[0].Content, StringComparison.Ordinal); + Assert.DoesNotContain("secret-plain", entry.Content[0].Content, StringComparison.Ordinal); + + foreach (var path in Directory.EnumerateFiles(root, "*", SearchOption.AllDirectories)) + { + var disk = Encoding.UTF8.GetString(await File.ReadAllBytesAsync( + path, + TestContext.Current.CancellationToken)); + Assert.DoesNotContain("sensitive-plain", disk, StringComparison.Ordinal); + Assert.DoesNotContain("secret-plain", disk, StringComparison.Ordinal); + Assert.DoesNotContain("public-result", disk, StringComparison.Ordinal); + } + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public async Task Same_semantic_snapshot_is_idempotent_and_changed_snapshot_conflicts() + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + inbox.PrepareRun("run-live"); + var first = Capture(inbox, ValidArgumentsJson); + var reordered = Capture(inbox, ReorderedArgumentsJson); + Assert.Equal(first.PayloadDigest, reordered.PayloadDigest); + + var spool = new WorkflowSpoolStore(root).Open("run-live"); + var normalizer = SeedNormalizer(); + using var gate = new SemaphoreSlim(1, 1); + var submission = inbox.SubmitAsync( + first, + TimeSpan.FromSeconds(2), + TestContext.Current.CancellationToken); + await WaitForPendingAsync(inbox, "run-live"); + Assert.Equal( + 1, + await CodexObserverRuntime.DrainDiagnosticsOnceAsync( + inbox, + "run-live", + normalizer, + spool, + null, + gate, + TestContext.Current.CancellationToken)); + Assert.True((await submission).Recorded); + + var replay = await inbox.SubmitAsync( + reordered, + TimeSpan.FromSeconds(1), + TestContext.Current.CancellationToken); + Assert.True(replay.Recorded); + Assert.Single(spool.ReadAfter(0, 10)); + + using var changedDocument = JsonDocument.Parse( + ValidArgumentsJson.Replace("public-result", "changed-result", StringComparison.Ordinal)); + Assert.True(DiagnosticSnapshotCapture.TryCreate( + ActiveRun(), + changedDocument.RootElement, + inbox.Protector, + s_timestamp, + out var changed, + out _)); + var conflict = await inbox.SubmitAsync( + changed!, + TimeSpan.FromSeconds(1), + TestContext.Current.CancellationToken); + Assert.False(conflict.Recorded); + Assert.Equal("snapshot_conflict", conflict.Code); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public async Task New_run_generation_rejects_delayed_prior_run_submissions() + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + inbox.PrepareRun("run-live"); + var stale = Capture(inbox, ValidArgumentsJson); + + inbox.PrepareRun("run-next"); + var result = await inbox.SubmitAsync( + stale, + TimeSpan.FromMilliseconds(100), + TestContext.Current.CancellationToken); + + Assert.False(result.Recorded); + Assert.Equal("run_closing", result.Code); + Assert.Empty(inbox.ReadPending("run-live")); + Assert.Single(Directory.EnumerateFiles( + Path.Combine(root, "diagnostic-inbox"), + "*.lock", + SearchOption.TopDirectoryOnly)); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public async Task Unreadable_inbox_request_is_quarantined_without_blocking_the_run() + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + inbox.PrepareRun("run-live"); + var requestPath = Path.Combine( + root, + "diagnostic-inbox", + "unreadable.request.qyl"); + await File.WriteAllTextAsync( + requestPath, + "{not-json", + TestContext.Current.CancellationToken); + + Assert.Empty(inbox.ReadPending("run-live")); + Assert.False(File.Exists(requestPath)); + Assert.True(File.Exists(requestPath + ".corrupt")); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public async Task Ack_retry_does_not_duplicate_spool_or_telemetry_projection() + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + inbox.PrepareRun("run-live"); + var request = Capture(inbox, ValidArgumentsJson); + var submission = inbox.SubmitAsync( + request, + TimeSpan.FromSeconds(2), + TestContext.Current.CancellationToken); + await WaitForPendingAsync(inbox, "run-live"); + + var normalizer = SeedNormalizer(); + var batch = normalizer.NormalizeDiagnosticSnapshot(request, s_timestamp); + var workflowEvent = Assert.Single(batch.Events); + var spool = new WorkflowSpoolStore(root).Open("run-live"); + await spool.AppendAsync( + new WorkflowSpoolEntry(workflowEvent, [request.Content]), + TestContext.Current.CancellationToken); + normalizer.MarkDiagnosticSnapshotRecorded(request.SnapshotId); + + var projected = 0; + using var gate = new SemaphoreSlim(1, 1); + Assert.Equal( + 1, + await CodexObserverRuntime.DrainDiagnosticsOnceAsync( + inbox, + "run-live", + normalizer, + spool, + _ => projected++, + gate, + TestContext.Current.CancellationToken)); + Assert.True((await submission).Recorded); + Assert.Equal(0, projected); + Assert.Single(spool.ReadAfter(0, 10)); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Theory] + [InlineData("duplicate-variable", "duplicate_variable")] + [InlineData("bad-actual", "invalid_actual_variable")] + [InlineData("expression", "invalid_check")] + public void Invalid_dynamic_input_is_rejected_with_machine_codes(string fixture, string expectedCode) + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + var json = fixture switch + { + "duplicate-variable" => + """{"snapshotId":"snapshot_1","probeId":"probe_1","phase":"input","variables":[{"name":"x","classification":"public","value":1},{"name":"x","classification":"internal","value":2}]}""", + "bad-actual" => + """{"snapshotId":"snapshot_1","probeId":"probe_1","phase":"input","variables":[],"checks":[{"checkId":"check_1","operator":"exists","actual":"not valid"}]}""", + _ => + """{"snapshotId":"snapshot_1","probeId":"probe_1","phase":"input","variables":[{"name":"x","classification":"public","value":1}],"checks":[{"checkId":"check_1","operator":"equal","actual":"x","expression":"x == 1"}]}""" + }; + using var document = JsonDocument.Parse(json); + + Assert.False(DiagnosticSnapshotCapture.TryCreate( + ActiveRun(), + document.RootElement, + inbox.Protector, + s_timestamp, + out _, + out var error)); + Assert.Equal(expectedCode, error.Code); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Theory] + [InlineData("variables", "too_many_variables", "variables")] + [InlineData("checks", "too_many_checks", "checks")] + [InlineData("depth", "value_too_deep", "variables[0].value")] + [InlineData("value", "value_too_large", "variables[0].value")] + [InlineData("captured-payload", "payload_too_large", "arguments")] + public void Capture_enforces_runtime_bounds(string fixture, string expectedCode, string expectedField) + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + var variables = fixture switch + { + "variables" => string.Join( + ',', + Enumerable.Range(0, DiagnosticSnapshotCapture.MaxVariables + 1) + .Select(static index => + $"{{\"name\":\"v{index}\",\"classification\":\"public\",\"value\":{index}}}")), + "depth" => + "{\"name\":\"deep\",\"classification\":\"public\",\"value\":" + + new string('[', DiagnosticSnapshotCapture.MaxValueDepth + 1) + + "0" + + new string(']', DiagnosticSnapshotCapture.MaxValueDepth + 1) + + "}", + "value" => + "{\"name\":\"large\",\"classification\":\"public\",\"value\":" + + "\"" + + new string('x', DiagnosticSnapshotCapture.MaxValueBytes) + + "\"" + + "}", + "captured-payload" => string.Join( + ',', + Enumerable.Range(0, 5).Select(index => + $"{{\"name\":\"large{index}\",\"classification\":\"public\",\"value\":" + + "\"" + + new string((char)('a' + index), 15_000) + + "\"" + + "}")), + _ => "{\"name\":\"x\",\"classification\":\"public\",\"value\":1}" + }; + var checks = fixture == "checks" + ? string.Join( + ',', + Enumerable.Range(0, DiagnosticSnapshotCapture.MaxChecks + 1) + .Select(static index => + $"{{\"checkId\":\"c{index}\",\"operator\":\"exists\",\"actual\":\"x\"}}")) + : ""; + var json = + "{\"snapshotId\":\"snapshot_bounds\",\"probeId\":\"probe_bounds\",\"phase\":\"input\",\"variables\":[" + + variables + + "]" + + (fixture == "checks" ? ",\"checks\":[" + checks + "]" : "") + + "}"; + using var document = JsonDocument.Parse(json); + + Assert.False(DiagnosticSnapshotCapture.TryCreate( + ActiveRun(), + document.RootElement, + inbox.Protector, + s_timestamp, + out _, + out var error)); + Assert.Equal(expectedCode, error.Code); + Assert.Equal(expectedField, error.Field); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public void Missing_operands_and_incompatible_types_are_unknown_without_expressions() + { + var root = TemporaryDirectory(); + try + { + var inbox = new DiagnosticSnapshotInbox(root); + using var document = JsonDocument.Parse( + """ + { + "snapshotId":"snapshot_unknown", + "probeId":"probe_unknown", + "phase":"input", + "variables":[ + {"name":"text","classification":"public","value":"1"}, + {"name":"number","classification":"public","value":1} + ], + "checks":[ + {"checkId":"missing_operand","operator":"equal","actual":"missing","expected":"number"}, + {"checkId":"incompatible","operator":"not_equal","actual":"text","expected":"number"}, + {"checkId":"missing_exists","operator":"exists","actual":"missing"} + ] + } + """); + + Assert.True(DiagnosticSnapshotCapture.TryCreate( + ActiveRun(), + document.RootElement, + inbox.Protector, + s_timestamp, + out var request, + out var error), error.Code); + Assert.Equal("fail", request!.Outcome); + Assert.Equal(1, request.FailedCheckCount); + using var captured = JsonDocument.Parse(request.Content.Content); + var outcomes = captured.RootElement.GetProperty("checks") + .EnumerateArray() + .ToDictionary( + static check => check.GetProperty("check_id").GetString()!, + static check => check.GetProperty("outcome").GetString()!, + StringComparer.Ordinal); + Assert.Equal("unknown", outcomes["missing_operand"]); + Assert.Equal("unknown", outcomes["incompatible"]); + Assert.Equal("fail", outcomes["missing_exists"]); + Assert.Equal(2, outcomes.Values.Count(static outcome => outcome == "unknown")); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + [Fact] + public void Telemetry_projection_emits_only_fixed_diagnostic_and_workflow_tags() + { + var root = TemporaryDirectory(); + Activity? turnActivity = null; + using var listener = new ActivityListener + { + ShouldListenTo = static source => source.Name == "qyl.codex.observer", + Sample = static (ref ActivityCreationOptions _) => + ActivitySamplingResult.AllDataAndRecorded, + ActivityStarted = activity => + { + if (activity.DisplayName == "codex.workflow.turn") + turnActivity = activity; + } + }; + ActivitySource.AddActivityListener(listener); + try + { + using var telemetry = WorkflowTelemetryProjection.Create("run-live", null); + var normalizer = new CodexEventNormalizer(); + Record(telemetry, normalizer.StartRun(s_timestamp)); + using (var thread = JsonDocument.Parse( + """{"method":"thread/started","params":{"thread":{"id":"thread-root","createdAt":1786276800}}}""")) + { + Record(telemetry, normalizer.Normalize(thread.RootElement, s_timestamp)); + } + using (var turn = JsonDocument.Parse( + """{"method":"turn/started","params":{"threadId":"thread-root","turn":{"id":"turn-root","startedAt":1786276800}}}""")) + { + Record(telemetry, normalizer.Normalize(turn.RootElement, s_timestamp)); + } + + var inbox = new DiagnosticSnapshotInbox(root); + var request = Capture(inbox, ValidArgumentsJson); + Record(telemetry, normalizer.NormalizeDiagnosticSnapshot(request, s_timestamp)); + telemetry.Record(new WorkflowEventAppend + { + EventId = new WorkflowEventId("other-content-extension"), + SourceSequence = 100, + Timestamp = s_timestamp, + Kind = WorkflowJournalEventKind.ContentCaptured, + ThreadId = "thread-root", + TurnId = "turn-root", + AttemptId = new WorkflowAttemptId("turn-root"), + Data = new Dictionary(StringComparer.Ordinal) + { + ["extension_id"] = "qyl.other.extension", + ["format_version"] = 1 + } + }); + + Assert.NotNull(turnActivity); + var diagnostic = Assert.Single( + turnActivity.Events, + static item => item.Name == "qyl.agent.diagnostic.snapshot"); + var tags = diagnostic.Tags.ToDictionary( + static item => item.Key, + static item => item.Value, + StringComparer.Ordinal); + Assert.Equal( + [ + "qyl.agent.diagnostic.check.count", + "qyl.agent.diagnostic.check.failed_count", + "qyl.agent.diagnostic.extension.id", + "qyl.agent.diagnostic.format.version", + "qyl.agent.diagnostic.outcome", + "qyl.agent.diagnostic.phase", + "qyl.agent.diagnostic.probe.id", + "qyl.agent.diagnostic.snapshot.id", + "qyl.agent.diagnostic.variable.count", + "qyl.workflow.attempt.id", + "qyl.workflow.event.id", + "qyl.workflow.run.id" + ], + tags.Keys.Order(StringComparer.Ordinal)); + Assert.Equal("qyl.agent.diagnostic.snapshot", tags["qyl.agent.diagnostic.extension.id"]); + Assert.Equal(1L, tags["qyl.agent.diagnostic.format.version"]); + Assert.Equal(6L, tags["qyl.agent.diagnostic.variable.count"]); + Assert.Equal(4L, tags["qyl.agent.diagnostic.check.count"]); + Assert.Equal(1L, tags["qyl.agent.diagnostic.check.failed_count"]); + Assert.DoesNotContain(tags, static item => + item.Key.Contains("result", StringComparison.Ordinal) || + item.Value?.ToString()?.Contains("sensitive-plain", StringComparison.Ordinal) is true || + item.Value?.ToString()?.Contains("secret-plain", StringComparison.Ordinal) is true); + } + finally + { + Directory.Delete(root, recursive: true); + } + } + + private static DiagnosticSnapshotInboxRequest Capture( + DiagnosticSnapshotInbox inbox, + string json) + { + using var document = JsonDocument.Parse(json); + Assert.True(DiagnosticSnapshotCapture.TryCreate( + ActiveRun(), + document.RootElement, + inbox.Protector, + s_timestamp, + out var request, + out var error), error.Code); + return request!; + } + + private static void Record( + WorkflowTelemetryProjection telemetry, + CodexNormalizedBatch batch) + { + foreach (var workflowEvent in batch.Events ?? []) + telemetry.Record(workflowEvent); + } + + private static CodexEventNormalizer SeedNormalizer() + { + var normalizer = new CodexEventNormalizer(); + normalizer.StartRun(s_timestamp); + using (var thread = JsonDocument.Parse( + """{"method":"thread/started","params":{"thread":{"id":"thread-root","createdAt":1786276800}}}""")) + { + normalizer.Normalize(thread.RootElement, s_timestamp); + } + using (var turn = JsonDocument.Parse( + """{"method":"turn/started","params":{"threadId":"thread-root","turn":{"id":"turn-root","startedAt":1786276800}}}""")) + { + normalizer.Normalize(turn.RootElement, s_timestamp); + } + return normalizer; + } + + private static async Task WaitForPendingAsync(DiagnosticSnapshotInbox inbox, string runId) + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2)); + while (inbox.ReadPending(runId).Count == 0) + await Task.Delay(10, timeout.Token); + } + + private static ActiveWorkflowRun ActiveRun() => + new("run-live", "thread-root", s_timestamp, Environment.ProcessId); + + private static string TemporaryDirectory() + { + var path = Path.Combine(Path.GetTempPath(), $"qyl-diagnostic-{Guid.NewGuid():N}"); + Directory.CreateDirectory(path); + return path; + } + + private const string ValidArgumentsJson = + """ + { + "snapshotId":"snapshot_1", + "probeId":"probe.output", + "phase":"checkpoint", + "variables":[ + {"name":"result","classification":"public","value":5}, + {"name":"label","classification":"public","value":"public-result"}, + {"name":"limit","classification":"internal","value":3}, + {"name":"token","classification":"sensitive","value":"sensitive-plain"}, + {"name":"password","classification":"secret","value":"secret-plain"}, + {"name":"missing","classification":"public","value":null} + ], + "checks":[ + {"checkId":"check_gt","operator":"greater_than","actual":"result","expected":"limit"}, + {"checkId":"check_type","operator":"type_is","actual":"result","expectedType":"integer"}, + {"checkId":"check_token","operator":"contains","actual":"token","expected":"token"}, + {"checkId":"check_exists","operator":"exists","actual":"missing"} + ] + } + """; + + private const string ReorderedArgumentsJson = + """ + { + "probeId":"probe.output", + "snapshotId":"snapshot_1", + "variables":[ + {"classification":"secret","value":"secret-plain","name":"password"}, + {"classification":"public","value":null,"name":"missing"}, + {"classification":"internal","value":3,"name":"limit"}, + {"classification":"sensitive","value":"sensitive-plain","name":"token"}, + {"classification":"public","value":5,"name":"result"} + ,{"classification":"public","value":"public-result","name":"label"} + ], + "phase":"checkpoint", + "checks":[ + {"actual":"missing","operator":"exists","checkId":"check_exists"}, + {"expected":"token","actual":"token","operator":"contains","checkId":"check_token"}, + {"expectedType":"integer","actual":"result","operator":"type_is","checkId":"check_type"}, + {"expected":"limit","actual":"result","operator":"greater_than","checkId":"check_gt"} + ] + } + """; +} diff --git a/tests/Qyl.Collector.Tests/AiDiagnosticSpanEventPersistenceTests.cs b/tests/Qyl.Collector.Tests/AiDiagnosticSpanEventPersistenceTests.cs new file mode 100644 index 00000000..f6ae442d --- /dev/null +++ b/tests/Qyl.Collector.Tests/AiDiagnosticSpanEventPersistenceTests.cs @@ -0,0 +1,130 @@ +using System.Text.Json; +using Google.Protobuf; +using OpenTelemetry.Proto.Collector.Trace.V1; +using OpenTelemetry.Proto.Common.V1; +using OpenTelemetry.Proto.Resource.V1; +using OpenTelemetry.Proto.Trace.V1; +using Qyl.Collector.Ingestion; +using Qyl.Collector.Storage; + +namespace Qyl.Collector.Tests; + +public sealed class AiDiagnosticSpanEventPersistenceTests +{ + private static readonly string[] s_fixedAttributeKeys = + [ + "qyl.agent.diagnostic.check.count", + "qyl.agent.diagnostic.check.failed_count", + "qyl.agent.diagnostic.extension.id", + "qyl.agent.diagnostic.format.version", + "qyl.agent.diagnostic.outcome", + "qyl.agent.diagnostic.phase", + "qyl.agent.diagnostic.probe.id", + "qyl.agent.diagnostic.snapshot.id", + "qyl.agent.diagnostic.variable.count", + "qyl.workflow.agent.id", + "qyl.workflow.attempt.id", + "qyl.workflow.event.id", + "qyl.workflow.run.id", + "qyl.workflow.tool_call.id" + ]; + + [Fact] + public async Task Fixed_diagnostic_projection_persists_on_span_events_without_dynamic_or_sensitive_payloads() + { + var diagnosticEvent = new Span.Types.Event + { + Name = "qyl.agent.diagnostic.snapshot", + TimeUnixNano = 2 + }; + diagnosticEvent.Attributes.Add( + [ + StringAttribute("qyl.agent.diagnostic.extension.id", "qyl.agent.diagnostic.snapshot"), + IntAttribute("qyl.agent.diagnostic.format.version", 1), + StringAttribute("qyl.agent.diagnostic.snapshot.id", "snapshot-1"), + StringAttribute("qyl.agent.diagnostic.probe.id", "probe-1"), + StringAttribute("qyl.agent.diagnostic.phase", "checkpoint"), + StringAttribute("qyl.agent.diagnostic.outcome", "fail"), + IntAttribute("qyl.agent.diagnostic.variable.count", 6), + IntAttribute("qyl.agent.diagnostic.check.count", 4), + IntAttribute("qyl.agent.diagnostic.check.failed_count", 1), + StringAttribute("qyl.workflow.run.id", "run-1"), + StringAttribute("qyl.workflow.event.id", "event-1"), + StringAttribute("qyl.workflow.attempt.id", "attempt-1"), + StringAttribute("qyl.workflow.agent.id", "agent-1"), + StringAttribute("qyl.workflow.tool_call.id", "tool-call-1"), + StringAttribute("qyl.agent.diagnostic.variable.connection_string", "dynamic-variable-value"), + StringAttribute("qyl.agent.diagnostic.variable.0.value", "dynamic-variable-payload"), + StringAttribute("qyl.agent.diagnostic.check.0.result", "dynamic-check-result"), + StringAttribute("qyl.agent.diagnostic.snapshot.secret", "secret-payload"), + StringAttribute("baggage.qyl.agent.diagnostic.snapshot.id", "baggage-payload") + ]); + + var span = new Span + { + TraceId = ByteString.CopyFrom(new byte[16]), + SpanId = ByteString.CopyFrom(new byte[8]), + Name = "codex.workflow.turn", + StartTimeUnixNano = 1, + EndTimeUnixNano = 3, + Events = { diagnosticEvent } + }; + var request = new ExportTraceServiceRequest + { + ResourceSpans = + { + new ResourceSpans + { + Resource = new Resource(), + ScopeSpans = { new ScopeSpans { Spans = { span } } } + } + } + }; + + var rows = IngestionStorageMapper.ToSpanStorageRows(OtlpConverter.ConvertTraceRequest(request)); + await using var store = new DuckDbStore(":memory:"); + await store.EnqueueAsync(new SpanBatch(rows), TestContext.Current.CancellationToken); + + var storedSpan = Assert.Single(await store.GetSpansAsync( + "default", + ct: TestContext.Current.CancellationToken)); + var storedEvent = Assert.Single(Assert.IsType>( + SpanChildStorage.DeserializeEvents(storedSpan.EventsJson))); + Assert.Equal("qyl.agent.diagnostic.snapshot", storedEvent.Name); + var attributesJson = Assert.IsType(storedEvent.AttributesJson); + using var attributes = JsonDocument.Parse(attributesJson); + + Assert.Equal( + s_fixedAttributeKeys, + attributes.RootElement.EnumerateObject().Select(static property => property.Name).ToArray()); + Assert.Equal( + "qyl.agent.diagnostic.snapshot", + attributes.RootElement.GetProperty("qyl.agent.diagnostic.extension.id").GetString()); + Assert.Equal( + "1", + attributes.RootElement.GetProperty("qyl.agent.diagnostic.format.version") + .GetProperty("value").GetString()); + Assert.DoesNotContain("dynamic-variable", attributesJson, StringComparison.Ordinal); + Assert.DoesNotContain("dynamic-check", attributesJson, StringComparison.Ordinal); + Assert.DoesNotContain("secret-payload", attributesJson, StringComparison.Ordinal); + Assert.DoesNotContain("baggage-payload", attributesJson, StringComparison.Ordinal); + } + + [Theory] + [InlineData("qyl.agent.diagnostic.variable.connection_string")] + [InlineData("qyl.agent.diagnostic.variable.0.value")] + [InlineData("qyl.agent.diagnostic.check.0.result")] + [InlineData("qyl.agent.diagnostic.snapshot.secret")] + [InlineData("baggage.qyl.agent.diagnostic.snapshot.id")] + public void Dynamic_and_sensitive_diagnostic_keys_are_not_captured(string key) + { + Assert.False(AttributeKeySets.ShouldCaptureSpanAttribute(key)); + Assert.False(AttributeKeySets.IsSafeSpanAttribute(key)); + } + + private static KeyValue StringAttribute(string key, string value) => + new() { Key = key, Value = new AnyValue { StringValue = value } }; + + private static KeyValue IntAttribute(string key, long value) => + new() { Key = key, Value = new AnyValue { IntValue = value } }; +} From 8f24ecfb6758e40162fc2c108e9e2b057b34e660 Mon Sep 17 00:00:00 2001 From: ancplua Date: Sun, 9 Aug 2026 04:16:18 +0200 Subject: [PATCH 2/2] Fix qyl verification gates --- .../qyl.collector/Storage/DuckDbStore.Retention.cs | 5 +++-- .../qyl.collector/Storage/DuckDbStore.Workflow.cs | 5 +++-- services/qyl.dashboard/package-lock.json | 12 ++++++------ 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/services/qyl.collector/Storage/DuckDbStore.Retention.cs b/services/qyl.collector/Storage/DuckDbStore.Retention.cs index 709d5266..03fb7065 100644 --- a/services/qyl.collector/Storage/DuckDbStore.Retention.cs +++ b/services/qyl.collector/Storage/DuckDbStore.Retention.cs @@ -61,8 +61,9 @@ public async Task DeleteExpiredWorkflowDataBatchAsync( await using (var select = con.CreateCommand()) { select.Transaction = transaction; - select.CommandText = """ - SELECT workflow_runs.project_id, workflow_runs.run_id + select.CommandText = $""" + SELECT workflow_runs.{WorkflowRunDbRow.ProjectIdColumnName}, + workflow_runs.{WorkflowRunDbRow.RunIdColumnName} FROM workflow_runs JOIN workflow_run_summaries AS summary ON summary.project_id = workflow_runs.project_id diff --git a/services/qyl.collector/Storage/DuckDbStore.Workflow.cs b/services/qyl.collector/Storage/DuckDbStore.Workflow.cs index b8189eba..dcb1595b 100644 --- a/services/qyl.collector/Storage/DuckDbStore.Workflow.cs +++ b/services/qyl.collector/Storage/DuckDbStore.Workflow.cs @@ -277,8 +277,9 @@ public Task> ListWorkflowRunsAsync( using var transaction = con.BeginTransaction(); using var command = con.CreateCommand(); command.Transaction = transaction; - command.CommandText = """ - SELECT workflow_runs.project_id, workflow_runs.run_id + command.CommandText = $""" + SELECT workflow_runs.{WorkflowRunDbRow.ProjectIdColumnName}, + workflow_runs.{WorkflowRunDbRow.RunIdColumnName} FROM workflow_runs AS workflow_runs JOIN workflow_run_summaries AS summary ON summary.project_id = workflow_runs.project_id diff --git a/services/qyl.dashboard/package-lock.json b/services/qyl.dashboard/package-lock.json index 61aec97c..dd9f95c0 100644 --- a/services/qyl.dashboard/package-lock.json +++ b/services/qyl.dashboard/package-lock.json @@ -2339,9 +2339,9 @@ "license": "MIT" }, "node_modules/fast-uri": { - "version": "3.1.4", - "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.4.tgz", - "integrity": "sha512-8JnbkQ4juDyvYs4mgFGQqg4yCYtFDtUtmp2QIQq11ZZe5CFQ5wcqm1rqDgAh/QdMySuBnPzMUiJUNZG5N/AiQw==", + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.5.tgz", + "integrity": "sha512-gHwA1O9LDIcKunMKhObS/HimwtehO1nPUECKAu5TpKgaO19fcWEl4bliWe1jWxVFvIXztJjjQ4L8XQ1EU9f7Jw==", "funding": [ { "type": "github", @@ -2872,9 +2872,9 @@ "license": "CC0-1.0" }, "node_modules/nanoid": { - "version": "3.3.16", - "resolved": "https://registry.npmjs.org/nanoid/-/nanoid-3.3.16.tgz", - "integrity": "sha512-bzlKTyNJ7+LdGIIwy8ijFpIqEQIvafahV7eYykJ8Cvh42EdJeODoJ6gUJXpQJvej1BddH8OqTXZNE/KfbWAu8Q==", + "version": "3.3.18", + "resolved": "https://registry.npmjs.org/nanoid/-/nanoid-3.3.18.tgz", + "integrity": "sha512-DTg4MJbGMWkfi6VZFdNt2/caMbQy4Ou+Op/hJQvGEWcnVfoA1QA+xzRKAzw9jD6+GVOOeYr/mIcuDSdug6F6+w==", "dev": true, "funding": [ {