Add AquaSynth live instrument daemon sessions

docs/daemonic-swarm-prior-work.md

@@ -12,9 +12,11 @@ architecture. It is a reference ledger for
   typed receipts. It now exposes `cultnet-host`, `cultnet-once`,
   `cultnet-stream`, and `cultnet-daemon` as local CultNet command surfaces.
 - `src/AquaSynth.Faust/AquaSynthDaemonService.cs`: the service owns
-  `patch.compile`, `instrument.sample`, and bounded `instrument.stream`
-  execution. It writes compile/session/render/stream/operator witnesses and
-  emits CultMesh-compatible stream descriptors and packet receipts.
+  `patch.compile`, `instrument.sample`, bounded `instrument.stream`,
+  `instrument.open`, `instrument.control`, and `instrument.block` execution. It
+  writes compile/session/render/stream/live/operator witnesses, keeps native
+  streaming patch instances for live sessions, and emits CultMesh-compatible
+  stream descriptors and packet receipts.
 - `src/AquaSynth.Faust/AquaSynthCultNetDaemon.cs`: local CultNet database host
   and command watcher. Typed command documents enter the database; typed
   receipts and provider state come back out.

docs/gamecult-daemonic-swarm.md

@@ -110,11 +110,13 @@ A good GameCult daemon publishes:
 
 AquaSynth is the current synth example. AquaSynth owns patch parsing, Faust
 emission, native compilation, compiled instrument sessions, sample rendering,
-bounded stream receipts, and synth-control state. A client can send
-`instrument.sample` or `instrument.stream` as typed command state through the
-local CultNet database. AquaSynth compiles or opens the instrument, writes
-sample artifacts and receipts, and publishes the result. The client consumes
-the capability; it does not become the synth.
+bounded stream receipts, live streaming sessions, and synth-control state. A
+client can send `instrument.sample`, `instrument.stream`, `instrument.open`,
+`instrument.control`, or `instrument.block` as typed command state through the
+local CultNet database. AquaSynth compiles or opens the instrument, keeps the
+native streaming patch instance, applies controls, writes sample/block artifacts
+and receipts, and publishes the result. The client consumes the capability; it
+does not become the synth.
 
 That lets one AquaSynth daemon sit on this machine or another machine in the
 Verse and serve multiple apps. The same engine can support Fensalir, Eve
@@ -287,12 +289,14 @@ The split is:
 - domain work inside the provider daemon.
 
 For AquaSynth, that means patch requests, compile receipts, session state, and
-stream descriptors are typed documents. Audio blocks should eventually move
-through a low-latency local transport such as shared-memory rings, platform
-audio graph nodes, memory-mapped packet buffers, or another fixed-block lane
-that does not drag filesystem writes or chatty command protocols into the audio
-callback. The command surface still owns intent and receipts. The realtime lane
-just moves sound fast.
+stream descriptors are typed documents. Live sessions already let a daemon open
+one native streaming patch and apply controls across fixed blocks; audio blocks
+should eventually move through a low-latency local transport such as
+shared-memory rings, platform audio graph nodes, memory-mapped packet buffers,
+or another fixed-block lane that does not drag filesystem writes or chatty
+command protocols into the audio callback. The command surface still owns
+intent, patch lifetime, controls, and receipts. The realtime lane just moves
+sound fast.
 
 That distinction matters. Transport is not authority. A ring buffer should not
 decide what instrument exists. A renderer should not decide which command

src/AquaSynth.Faust/AquaSynthCultNetDaemon.cs

@@ -108,6 +108,42 @@ public sealed class AquaSynthCultNetDaemon : IAsyncDisposable
             timeout);
     }
 
+    public Task<AquaSynthInstrumentOpenReceipt> SubmitOpenAsync(
+        AquaSynthInstrumentOpenCommand command,
+        TimeSpan? timeout = null)
+    {
+        var key = OpenReceiptKey(command.CommandId);
+        return SubmitAndWaitAsync<AquaSynthInstrumentOpenCommand, AquaSynthInstrumentOpenReceipt>(
+            command,
+            CommandKey(command.CommandId),
+            key,
+            timeout);
+    }
+
+    public Task<AquaSynthInstrumentControlReceipt> SubmitControlAsync(
+        AquaSynthInstrumentControlCommand command,
+        TimeSpan? timeout = null)
+    {
+        var key = ControlReceiptKey(command.CommandId);
+        return SubmitAndWaitAsync<AquaSynthInstrumentControlCommand, AquaSynthInstrumentControlReceipt>(
+            command,
+            CommandKey(command.CommandId),
+            key,
+            timeout);
+    }
+
+    public Task<AquaSynthInstrumentBlockReceipt> SubmitBlockAsync(
+        AquaSynthInstrumentBlockCommand command,
+        TimeSpan? timeout = null)
+    {
+        var key = BlockReceiptKey(command.CommandId);
+        return SubmitAndWaitAsync<AquaSynthInstrumentBlockCommand, AquaSynthInstrumentBlockReceipt>(
+            command,
+            CommandKey(command.CommandId),
+            key,
+            timeout);
+    }
+
     public async ValueTask DisposeAsync()
     {
         foreach (var subscription in subscriptions)
@@ -183,6 +219,27 @@ public sealed class AquaSynthCultNetDaemon : IAsyncDisposable
                 _ = Task.Run(() => HandleStreamAsync(command));
             }
         }));
+        subscriptions.Add(database.Watch<AquaSynthInstrumentOpenCommand>().Subscribe(change =>
+        {
+            if (change.Document is { } command && ShouldHandle(nameof(AquaSynthInstrumentOpenCommand), command.CommandId))
+            {
+                _ = Task.Run(() => HandleOpenAsync(command));
+            }
+        }));
+        subscriptions.Add(database.Watch<AquaSynthInstrumentControlCommand>().Subscribe(change =>
+        {
+            if (change.Document is { } command && ShouldHandle(nameof(AquaSynthInstrumentControlCommand), command.CommandId))
+            {
+                _ = Task.Run(() => HandleControlAsync(command));
+            }
+        }));
+        subscriptions.Add(database.Watch<AquaSynthInstrumentBlockCommand>().Subscribe(change =>
+        {
+            if (change.Document is { } command && ShouldHandle(nameof(AquaSynthInstrumentBlockCommand), command.CommandId))
+            {
+                _ = Task.Run(() => HandleBlockAsync(command));
+            }
+        }));
     }
 
     private bool ShouldHandle(string commandType, string commandId)
@@ -239,6 +296,51 @@ public sealed class AquaSynthCultNetDaemon : IAsyncDisposable
         }
     }
 
+    private async Task HandleOpenAsync(AquaSynthInstrumentOpenCommand command)
+    {
+        await commandGate.WaitAsync().ConfigureAwait(false);
+        try
+        {
+            var receipt = await service.OpenInstrumentAsync(command).ConfigureAwait(false);
+            await Database.PutAsync(OpenReceiptKey(command.CommandId), receipt).ConfigureAwait(false);
+            await PublishProviderStateAsync().ConfigureAwait(false);
+        }
+        finally
+        {
+            commandGate.Release();
+        }
+    }
+
+    private async Task HandleControlAsync(AquaSynthInstrumentControlCommand command)
+    {
+        await commandGate.WaitAsync().ConfigureAwait(false);
+        try
+        {
+            var receipt = await service.ControlInstrumentAsync(command).ConfigureAwait(false);
+            await Database.PutAsync(ControlReceiptKey(command.CommandId), receipt).ConfigureAwait(false);
+            await PublishProviderStateAsync().ConfigureAwait(false);
+        }
+        finally
+        {
+            commandGate.Release();
+        }
+    }
+
+    private async Task HandleBlockAsync(AquaSynthInstrumentBlockCommand command)
+    {
+        await commandGate.WaitAsync().ConfigureAwait(false);
+        try
+        {
+            var receipt = await service.ProcessInstrumentBlockAsync(command).ConfigureAwait(false);
+            await Database.PutAsync(BlockReceiptKey(command.CommandId), receipt).ConfigureAwait(false);
+            await PublishProviderStateAsync().ConfigureAwait(false);
+        }
+        finally
+        {
+            commandGate.Release();
+        }
+    }
+
     private Task PublishProviderStateAsync()
     {
         var state = new AquaSynthCultNetProviderState(
@@ -250,12 +352,18 @@ public sealed class AquaSynthCultNetDaemon : IAsyncDisposable
             [
                 AquaSynthDaemonSchemas.PatchCompileCommand,
                 AquaSynthDaemonSchemas.InstrumentSampleCommand,
-                AquaSynthDaemonSchemas.AutomationStreamCommand
+                AquaSynthDaemonSchemas.AutomationStreamCommand,
+                AquaSynthDaemonSchemas.InstrumentOpenCommand,
+                AquaSynthDaemonSchemas.InstrumentControlCommand,
+                AquaSynthDaemonSchemas.InstrumentBlockCommand
             ],
             [
                 AquaSynthDaemonSchemas.PatchCompileReceipt,
                 AquaSynthDaemonSchemas.RenderSampleReceipt,
-                AquaSynthDaemonSchemas.AutomationStreamReceipt
+                AquaSynthDaemonSchemas.AutomationStreamReceipt,
+                AquaSynthDaemonSchemas.InstrumentOpenReceipt,
+                AquaSynthDaemonSchemas.InstrumentControlReceipt,
+                AquaSynthDaemonSchemas.InstrumentBlockReceipt
             ],
             [
                 "global:aquasynth.cultnet_provider",
@@ -271,6 +379,12 @@ public sealed class AquaSynthCultNetDaemon : IAsyncDisposable
     public static CultRecordKey RenderReceiptKey(string commandId) => new($"{commandId}-render");
 
     public static CultRecordKey StreamReceiptKey(string commandId) => new($"{commandId}-stream");
+
+    public static CultRecordKey OpenReceiptKey(string commandId) => new($"{commandId}-open");
+
+    public static CultRecordKey ControlReceiptKey(string commandId) => new($"{commandId}-control");
+
+    public static CultRecordKey BlockReceiptKey(string commandId) => new($"{commandId}-block");
 }
 
 public static class AquaSynthCultNetDocuments
@@ -280,10 +394,16 @@ public static class AquaSynthCultNetDocuments
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthPatchCompileCommand>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthInstrumentSampleCommand>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthAutomationStreamCommand>())
+            .Register(CultNetDocumentBinding.ForDocument<AquaSynthInstrumentOpenCommand>())
+            .Register(CultNetDocumentBinding.ForDocument<AquaSynthInstrumentControlCommand>())
+            .Register(CultNetDocumentBinding.ForDocument<AquaSynthInstrumentBlockCommand>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthCompiledInstrumentSession>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthPatchCompileReceipt>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthRenderSampleReceipt>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthAutomationStreamReceipt>())
+            .Register(CultNetDocumentBinding.ForDocument<AquaSynthInstrumentOpenReceipt>())
+            .Register(CultNetDocumentBinding.ForDocument<AquaSynthInstrumentControlReceipt>())
+            .Register(CultNetDocumentBinding.ForDocument<AquaSynthInstrumentBlockReceipt>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthOperatorState>())
             .Register(CultNetDocumentBinding.ForDocument<AquaSynthCultNetProviderState>());
 }

src/AquaSynth.Faust/AquaSynthDaemonService.cs

@@ -17,6 +17,12 @@ public static class AquaSynthDaemonSchemas
     public const string RenderSampleReceipt = "aquasynth.render_sample_receipt.v1";
     public const string AutomationStreamCommand = "aquasynth.command.automation_stream.v1";
     public const string AutomationStreamReceipt = "aquasynth.automation_stream_receipt.v1";
+    public const string InstrumentOpenCommand = "aquasynth.command.instrument_open.v1";
+    public const string InstrumentControlCommand = "aquasynth.command.instrument_control.v1";
+    public const string InstrumentBlockCommand = "aquasynth.command.instrument_block.v1";
+    public const string InstrumentOpenReceipt = "aquasynth.instrument_open_receipt.v1";
+    public const string InstrumentControlReceipt = "aquasynth.instrument_control_receipt.v1";
+    public const string InstrumentBlockReceipt = "aquasynth.instrument_block_receipt.v1";
     public const string OperatorState = "aquasynth.operator_state.v1";
 }
 
@@ -70,6 +76,34 @@ public sealed record AquaSynthAutomationControlFrame(
     [property: Key(0)] int Block,
     [property: Key(1)] Dictionary<string, float> Controls);
 
+[CultDocument("aquasynth.command.instrument_open", AquaSynthDaemonSchemas.InstrumentOpenCommand)]
+[MessagePackObject]
+public sealed record AquaSynthInstrumentOpenCommand(
+    [property: Key(0), CultName] string CommandId,
+    [property: Key(1)] string PatchId,
+    [property: Key(2)] string FaustName,
+    [property: Key(3)] string Script,
+    [property: Key(4)] int BlockSize = 128,
+    [property: Key(5)] float Gain = 1.0f,
+    [property: Key(6)] Dictionary<string, float>? Controls = null,
+    [property: Key(7)] int Revision = 1);
+
+[CultDocument("aquasynth.command.instrument_control", AquaSynthDaemonSchemas.InstrumentControlCommand)]
+[MessagePackObject]
+public sealed record AquaSynthInstrumentControlCommand(
+    [property: Key(0), CultName] string CommandId,
+    [property: Key(1), CultIndex("session")] string SessionId,
+    [property: Key(2)] Dictionary<string, float> Controls);
+
+[CultDocument("aquasynth.command.instrument_block", AquaSynthDaemonSchemas.InstrumentBlockCommand)]
+[MessagePackObject]
+public sealed record AquaSynthInstrumentBlockCommand(
+    [property: Key(0), CultName] string CommandId,
+    [property: Key(1), CultIndex("session")] string SessionId,
+    [property: Key(2)] int FrameCount = 128,
+    [property: Key(3)] float Gain = 1.0f,
+    [property: Key(4)] Dictionary<string, float>? Controls = null);
+
 [CultDocument("aquasynth.compiled_instrument_session", AquaSynthDaemonSchemas.CompiledInstrumentSession)]
 [MessagePackObject]
 public sealed record AquaSynthCompiledInstrumentSession(
@@ -143,6 +177,54 @@ public sealed record AquaSynthAutomationStreamReceipt(
     [property: Key(14)] string FailureCode = "",
     [property: Key(15)] string FailureMessage = "");
 
+[CultDocument("aquasynth.instrument_open_receipt", AquaSynthDaemonSchemas.InstrumentOpenReceipt)]
+[MessagePackObject]
+public sealed record AquaSynthInstrumentOpenReceipt(
+    [property: Key(0), CultName] string ReceiptId,
+    [property: Key(1), CultIndex("command")] string CommandId,
+    [property: Key(2), CultIndex("patch")] string PatchId,
+    [property: Key(3), CultIndex("session")] string SessionId,
+    [property: Key(4)] string Status,
+    [property: Key(5)] string OpenedAtUtc,
+    [property: Key(6)] int SampleRate,
+    [property: Key(7)] int InputCount,
+    [property: Key(8)] int OutputCount,
+    [property: Key(9)] int BlockSize,
+    [property: Key(10)] string[] ControlPaths,
+    [property: Key(11)] string[] ProbePaths,
+    [property: Key(12)] string FailureCode = "",
+    [property: Key(13)] string FailureMessage = "");
+
+[CultDocument("aquasynth.instrument_control_receipt", AquaSynthDaemonSchemas.InstrumentControlReceipt)]
+[MessagePackObject]
+public sealed record AquaSynthInstrumentControlReceipt(
+    [property: Key(0), CultName] string ReceiptId,
+    [property: Key(1), CultIndex("command")] string CommandId,
+    [property: Key(2), CultIndex("session")] string SessionId,
+    [property: Key(3)] string Status,
+    [property: Key(4)] string AppliedAtUtc,
+    [property: Key(5)] int ControlCount,
+    [property: Key(6)] string FailureCode = "",
+    [property: Key(7)] string FailureMessage = "");
+
+[CultDocument("aquasynth.instrument_block_receipt", AquaSynthDaemonSchemas.InstrumentBlockReceipt)]
+[MessagePackObject]
+public sealed record AquaSynthInstrumentBlockReceipt(
+    [property: Key(0), CultName] string BlockId,
+    [property: Key(1), CultIndex("command")] string CommandId,
+    [property: Key(2), CultIndex("session")] string SessionId,
+    [property: Key(3)] string Status,
+    [property: Key(4)] string CompletedAtUtc,
+    [property: Key(5)] ulong Sequence,
+    [property: Key(6)] int SampleRate,
+    [property: Key(7)] int SampleCount,
+    [property: Key(8)] float Peak,
+    [property: Key(9)] float Rms,
+    [property: Key(10)] string Float32Uri,
+    [property: Key(11)] string ContentHash,
+    [property: Key(12)] string FailureCode = "",
+    [property: Key(13)] string FailureMessage = "");
+
 [MessagePackObject]
 public sealed record AquaSynthCultMeshStreamDescriptor(
     [property: Key(0)] string StreamId,
@@ -192,6 +274,7 @@ public sealed class AquaSynthDaemonService : IDisposable
     private readonly AquaSynthDaemonOptions options;
     private readonly AquaSynthPatchCompiler compiler;
     private readonly Dictionary<string, AquaSynthCompiledPatch> sessions = new(StringComparer.Ordinal);
+    private readonly Dictionary<string, LiveInstrumentSession> liveSessions = new(StringComparer.Ordinal);
     private int compileCount;
     private int renderCount;
     private int streamCount;
@@ -238,6 +321,11 @@ public sealed class AquaSynthDaemonService : IDisposable
         }
 
         var sessionId = SessionId(command.PatchId, patch!.Manifest.CompileKey);
+        if (liveSessions.Remove(sessionId, out var live))
+        {
+            live.Dispose();
+        }
+
         if (sessions.Remove(sessionId, out var old))
         {
             old.Dispose();
@@ -279,6 +367,175 @@ public sealed class AquaSynthDaemonService : IDisposable
         return receipt;
     }
 
+    public async Task<AquaSynthInstrumentOpenReceipt> OpenInstrumentAsync(AquaSynthInstrumentOpenCommand command)
+    {
+        ObjectDisposedException.ThrowIf(disposed, this);
+        ValidateOpen(command);
+
+        var durationSeconds = command.BlockSize / (float)options.SampleRate;
+        var compile = await CompileAsync(new AquaSynthPatchCompileCommand(
+            command.CommandId,
+            command.PatchId,
+            command.FaustName,
+            command.Script,
+            command.Revision,
+            durationSeconds)).ConfigureAwait(false);
+
+        if (!string.Equals(compile.Status, "succeeded", StringComparison.Ordinal))
+        {
+            var failed = new AquaSynthInstrumentOpenReceipt(
+                ReceiptId(command.CommandId, "open"),
+                command.CommandId,
+                command.PatchId,
+                compile.SessionId,
+                "failed",
+                Now(),
+                options.SampleRate,
+                0,
+                0,
+                command.BlockSize,
+                [],
+                [],
+                "compile_failed",
+                compile.FailureMessage);
+            await WriteReceiptAsync("live", failed.ReceiptId, failed).ConfigureAwait(false);
+            await WriteOperatorStateAsync(command.CommandId, failed.Status).ConfigureAwait(false);
+            return failed;
+        }
+
+        var patch = sessions[compile.SessionId];
+        var controls = NormalizeControls(command.Controls) ?? new Dictionary<string, float>(StringComparer.OrdinalIgnoreCase);
+        if (liveSessions.Remove(compile.SessionId, out var oldLive))
+        {
+            oldLive.Dispose();
+        }
+
+        liveSessions[compile.SessionId] = new LiveInstrumentSession(
+            compile.SessionId,
+            command.PatchId,
+            command.BlockSize,
+            command.Gain,
+            patch,
+            patch.CreateStreamingPatch(),
+            controls);
+
+        var receipt = new AquaSynthInstrumentOpenReceipt(
+            ReceiptId(command.CommandId, "open"),
+            command.CommandId,
+            command.PatchId,
+            compile.SessionId,
+            "succeeded",
+            Now(),
+            patch.Manifest.SampleRate,
+            patch.Manifest.InputCount,
+            patch.Manifest.OutputCount,
+            command.BlockSize,
+            [.. patch.ControlPaths],
+            [.. patch.ProbePaths]);
+        await WriteReceiptAsync("live", receipt.ReceiptId, receipt).ConfigureAwait(false);
+        await WriteOperatorStateAsync(command.CommandId, receipt.Status).ConfigureAwait(false);
+        return receipt;
+    }
+
+    public async Task<AquaSynthInstrumentControlReceipt> ControlInstrumentAsync(AquaSynthInstrumentControlCommand command)
+    {
+        ObjectDisposedException.ThrowIf(disposed, this);
+        ValidateControl(command);
+
+        if (!liveSessions.TryGetValue(command.SessionId, out var session))
+        {
+            var missing = new AquaSynthInstrumentControlReceipt(
+                ReceiptId(command.CommandId, "control"),
+                command.CommandId,
+                command.SessionId,
+                "failed",
+                Now(),
+                0,
+                "session_not_found",
+                $"Live AquaSynth session '{command.SessionId}' is not open.");
+            await WriteReceiptAsync("live", missing.ReceiptId, missing).ConfigureAwait(false);
+            await WriteOperatorStateAsync(command.CommandId, missing.Status).ConfigureAwait(false);
+            return missing;
+        }
+
+        var controls = NormalizeControls(command.Controls) ?? [];
+        session.ApplyControls(controls);
+        var receipt = new AquaSynthInstrumentControlReceipt(
+            ReceiptId(command.CommandId, "control"),
+            command.CommandId,
+            command.SessionId,
+            "succeeded",
+            Now(),
+            controls.Count);
+        await WriteReceiptAsync("live", receipt.ReceiptId, receipt).ConfigureAwait(false);
+        await WriteOperatorStateAsync(command.CommandId, receipt.Status).ConfigureAwait(false);
+        return receipt;
+    }
+
+    public async Task<AquaSynthInstrumentBlockReceipt> ProcessInstrumentBlockAsync(AquaSynthInstrumentBlockCommand command)
+    {
+        ObjectDisposedException.ThrowIf(disposed, this);
+        ValidateBlock(command);
+
+        if (!liveSessions.TryGetValue(command.SessionId, out var session))
+        {
+            var missing = new AquaSynthInstrumentBlockReceipt(
+                BlockId(command.CommandId),
+                command.CommandId,
+                command.SessionId,
+                "failed",
+                Now(),
+                0,
+                options.SampleRate,
+                0,
+                0,
+                0,
+                "",
+                "",
+                "session_not_found",
+                $"Live AquaSynth session '{command.SessionId}' is not open.");
+            await WriteReceiptAsync("live", missing.BlockId, missing).ConfigureAwait(false);
+            await WriteOperatorStateAsync(command.CommandId, missing.Status).ConfigureAwait(false);
+            return missing;
+        }
+
+        if (NormalizeControls(command.Controls) is { } controls)
+        {
+            session.ApplyControls(controls);
+        }
+
+        var outputs = new float[Math.Max(1, session.Stream.OutputCount)][];
+        for (var channel = 0; channel < outputs.Length; channel++)
+        {
+            outputs[channel] = new float[command.FrameCount];
+        }
+
+        session.Stream.ProcessBlock([], outputs, command.FrameCount, session.Controls);
+        var mono = MixMono(outputs, command.FrameCount, command.Gain * session.Gain);
+        var blockId = BlockId(command.CommandId);
+        var blockDir = Path.Combine(options.StoreRoot, "live", SafeFileName(command.SessionId));
+        Directory.CreateDirectory(blockDir);
+        var floatPath = Path.Combine(blockDir, $"{SafeFileName(blockId)}.f32");
+        await WriteFloat32Async(floatPath, mono).ConfigureAwait(false);
+
+        var receipt = new AquaSynthInstrumentBlockReceipt(
+            blockId,
+            command.CommandId,
+            command.SessionId,
+            "succeeded",
+            Now(),
+            session.NextSequence(),
+            session.Patch.Manifest.SampleRate,
+            mono.Length,
+            Peak(mono),
+            Rms(mono),
+            ToArtifactUri(floatPath),
+            await Sha256Async(floatPath).ConfigureAwait(false));
+        await WriteReceiptAsync("live", receipt.BlockId, receipt).ConfigureAwait(false);
+        await WriteOperatorStateAsync(command.CommandId, receipt.Status).ConfigureAwait(false);
+        return receipt;
+    }
+
     public async Task<AquaSynthSampleResult> SampleAsync(AquaSynthInstrumentSampleCommand command)
     {
         ObjectDisposedException.ThrowIf(disposed, this);
@@ -493,11 +750,17 @@ public sealed class AquaSynthDaemonService : IDisposable
             return;
         }
 
+        foreach (var session in liveSessions.Values)
+        {
+            session.Dispose();
+        }
+
         foreach (var patch in sessions.Values)
         {
             patch.Dispose();
         }
 
+        liveSessions.Clear();
         sessions.Clear();
         compiler.Dispose();
         disposed = true;
@@ -520,6 +783,7 @@ public sealed class AquaSynthDaemonService : IDisposable
                 "aquasynth.instrument/sessions",
                 "aquasynth.render/samples",
                 "aquasynth.instrument/automation-streams",
+                "aquasynth.instrument/live-sessions",
                 "aquasynth.operator/status"
             ]);
         await WriteReceiptAsync("operator", "operator-state", state, global: true).ConfigureAwait(false);
@@ -598,6 +862,38 @@ public sealed class AquaSynthDaemonService : IDisposable
         }
     }
 
+    private static void ValidateOpen(AquaSynthInstrumentOpenCommand command)
+    {
+        Require(command.CommandId, nameof(command.CommandId));
+        Require(command.PatchId, nameof(command.PatchId));
+        Require(command.FaustName, nameof(command.FaustName));
+        Require(command.Script, nameof(command.Script));
+        if (command.BlockSize <= 0)
+        {
+            throw new ArgumentOutOfRangeException(nameof(command.BlockSize), "Live instrument block size must be positive.");
+        }
+    }
+
+    private static void ValidateControl(AquaSynthInstrumentControlCommand command)
+    {
+        Require(command.CommandId, nameof(command.CommandId));
+        Require(command.SessionId, nameof(command.SessionId));
+        if (command.Controls.Count == 0)
+        {
+            throw new ArgumentException("At least one control value is required.", nameof(command.Controls));
+        }
+    }
+
+    private static void ValidateBlock(AquaSynthInstrumentBlockCommand command)
+    {
+        Require(command.CommandId, nameof(command.CommandId));
+        Require(command.SessionId, nameof(command.SessionId));
+        if (command.FrameCount <= 0)
+        {
+            throw new ArgumentOutOfRangeException(nameof(command.FrameCount), "Live instrument frame count must be positive.");
+        }
+    }
+
     private static void Require(string value, string name)
     {
         if (string.IsNullOrWhiteSpace(value))
@@ -713,6 +1009,8 @@ public sealed class AquaSynthDaemonService : IDisposable
 
     private static string RenderId(string commandId) => $"{commandId}-render";
 
+    private static string BlockId(string commandId) => $"{commandId}-block";
+
     private static string SessionId(string patchId, string compileKey) => $"{SafeFileName(patchId)}-{compileKey[..Math.Min(12, compileKey.Length)]}";
 
     private static string SafeFileName(string value)
@@ -729,4 +1027,63 @@ public sealed class AquaSynthDaemonService : IDisposable
     private static string ToArtifactUri(string path) => $"file:///{Path.GetFullPath(path).Replace('\\', '/')}";
 
     private static string Now() => DateTimeOffset.UtcNow.ToString("O");
+
+    private sealed class LiveInstrumentSession : IDisposable
+    {
+        private ulong sequence;
+        private bool disposed;
+
+        public LiveInstrumentSession(
+            string sessionId,
+            string patchId,
+            int blockSize,
+            float gain,
+            AquaSynthCompiledPatch patch,
+            AquaSynthStreamingPatch stream,
+            Dictionary<string, float> controls)
+        {
+            SessionId = sessionId;
+            PatchId = patchId;
+            BlockSize = blockSize;
+            Gain = gain;
+            Patch = patch;
+            Stream = stream;
+            Controls = controls;
+        }
+
+        public string SessionId { get; }
+
+        public string PatchId { get; }
+
+        public int BlockSize { get; }
+
+        public float Gain { get; }
+
+        public AquaSynthCompiledPatch Patch { get; }
+
+        public AquaSynthStreamingPatch Stream { get; }
+
+        public Dictionary<string, float> Controls { get; }
+
+        public void ApplyControls(IReadOnlyDictionary<string, float> controls)
+        {
+            foreach (var pair in controls)
+            {
+                Controls[pair.Key] = pair.Value;
+            }
+        }
+
+        public ulong NextSequence() => sequence++;
+
+        public void Dispose()
+        {
+            if (disposed)
+            {
+                return;
+            }
+
+            Stream.Dispose();
+            disposed = true;
+        }
+    }
 }

state/evidence.jsonl

@@ -282,4 +282,5 @@
 {"timestamp":"2026-06-15T01:42:00+01:00","topic":"aquasynth-local-cultnet-daemon","claim":"AquaSynth now has a local CultNet daemon boundary: typed patch/sample/stream command documents are written to the CultNet database, AquaSynthCultNetDaemon watches and serializes them through AquaSynthDaemonService, and typed compile/render/stream/provider receipts are published back to the database. The service remains the render/session owner; CultNet is transport and publication authority.","evidence":"src/AquaSynth.Faust/AquaSynthCultNetDaemon.cs; tools/AquaSynthDaemon/Program.cs; tests/AquaSynth.Dsl.Tests/AquaSynthDaemonServiceTests.cs; docs/verse-service-contract.md; CLI smokes: cultnet-once returned 2205 sample frames for 0.05s and cultnet-stream returned 2 packet receipts with control frame count 1 on block 1."}
 {"timestamp":"2026-06-15T02:05:00+01:00","topic":"gamecult-daemonic-swarm-doc","claim":"GameCult app architecture is now documented as a daemonic swarm pattern: the visible window is usually a thin native Eve terminal, while provider daemons own capabilities, typed state, commands, receipts, health, and surfaces. The prior-work ledger ties the article back to AquaSynth, Eve, CultMesh, Odin, Idunn, and public GameCult site references.","evidence":"docs/gamecult-daemonic-swarm.md; docs/daemonic-swarm-prior-work.md; state/map.yaml"}
 {"timestamp":"2026-06-15T02:22:00+01:00","topic":"cinematic-daemonic-swarm-pipeline","claim":"The daemonic swarm documentation now includes the creative cinematic pipeline: Ghostlight coordinates scene state and next beats, Weksa lowers writing/agent state into utterance and delivery intent, AquaSynth or external providers produce voice, Vili wraps Kimodo for motion generation, and Brokkr mirrors/synchronizes Unity and Blender through CultMesh without stealing editor truth.","evidence":"docs/gamecult-daemonic-swarm.md; docs/daemonic-swarm-prior-work.md; Brokkr README/docs/provider-contract.md; Vili README/docs/verse-service-contract.md; Ghostlight prompts/sandboxed-coordinator-turn.md; E:/Projects/weksa/docs/verse-service-contract.md"}
+{"timestamp":"2026-06-15T03:04:00+01:00","topic":"aquasynth-daemon-live-session-control","claim":"AquaSynth live daemon control now belongs to AquaSynthDaemonService: instrument.open compiles once and keeps a native streaming patch instance, instrument.control updates session controls, and instrument.block processes fixed blocks through the same live session. AquaSynthCultNetDaemon only observes typed command documents and publishes typed receipts; it is not a second synth/session owner.","evidence":"src/AquaSynth.Faust/AquaSynthDaemonService.cs; src/AquaSynth.Faust/AquaSynthCultNetDaemon.cs; tools/AquaSynthDaemon/Program.cs; tests/AquaSynth.Dsl.Tests/AquaSynthDaemonServiceTests.cs; focused AquaSynthDaemonServiceTests passed 5/5"}
 {"timestamp":"2026-06-15T02:35:00+01:00","topic":"greater-daemon-composition-rule","claim":"The daemonic swarm docs now state the greater-daemon composition rule: smaller daemon capabilities compose upward through typed command/receipt surfaces, while larger workflow daemons own orchestration and aggregate receipts without absorbing lower-daemon truth, provenance, rejections, or failures.","evidence":"docs/gamecult-daemonic-swarm.md; docs/daemonic-swarm-prior-work.md"}

state/map.yaml

@@ -124,7 +124,7 @@ boundaries:
     - Faust source generation in AquaSynth.Core
     - target-code generation, compile-product contracts, dynamic patch compilation policy, cache identity, generated artifact manifests, and Faust toolchain selection in AquaSynth.Faust
     - live patch compilation, native Faust factory lifetime, and offline sample-buffer render proof in AquaSynth.Faust
-    - AquaSynth daemon command ownership for patch.compile, instrument.sample, and bounded instrument.stream automation: compile/session/render/stream/operator typed receipts, requested-duration sample renders, CultMesh-compatible audio/control stream descriptors, per-block .f32 packet artifacts, JSON-lines local transport through tools/AquaSynthDaemon, and a local CultNet host/database that watches typed command documents and publishes typed receipt/provider documents without becoming a second render/session authority
+    - AquaSynth daemon command ownership for patch.compile, instrument.sample, bounded instrument.stream automation, and live instrument.open/instrument.control/instrument.block sessions: compile/session/render/stream/live/operator typed receipts, requested-duration sample renders, persistent native streaming patch instances for live controls, CultMesh-compatible audio/control stream descriptors, per-block .f32 packet artifacts, JSON-lines local transport through tools/AquaSynthDaemon, and a local CultNet host/database that watches typed command documents and publishes typed receipt/provider documents without becoming a second render/session authority
     - IPA-to-phonetic-feature planning, articulatory gesture planning, and human/alien vocal-tract roadmap design
     - phonetic intent, articulatory plan, morphology capability, and constraint-report contracts for vocal tract synthesis
     - packed-buffer native backprop/Adam as the shared AquaSynth gradient-descent primitive

state/spine.yaml

@@ -57,7 +57,7 @@ current_mechanism:
 
 live_invariants:
   - AquaSynth owns its patch graph; external synths are references, not runtime dependencies.
-  - AquaSynthDaemonService owns compile/render/stream execution for patch.compile, instrument.sample, and bounded instrument.stream automation. It delegates DSP to AquaSynth.Faust, writes compile/session/render/stream/operator .cc witnesses under the configured .aquasynth store, writes .f32/.wav sample artifacts, and emits CultMesh-compatible audio/control stream descriptors plus per-block packet receipts. AquaSynthCultNetDaemon owns the local CultNet command-observation boundary: typed command documents enter the CultNet database, the daemon serializes execution through AquaSynthDaemonService, and typed receipt/provider documents come back out. JSON-lines modes are only operator/xeno edges; they must not become second audio/session authorities.
+  - AquaSynthDaemonService owns compile/render/stream/live-session execution for patch.compile, instrument.sample, bounded instrument.stream automation, instrument.open, instrument.control, and instrument.block. It delegates DSP to AquaSynth.Faust, writes compile/session/render/stream/live/operator .cc witnesses under the configured .aquasynth store, writes .f32/.wav sample artifacts, keeps native streaming patch instances for live sessions, and emits CultMesh-compatible audio/control stream descriptors plus per-block packet receipts. AquaSynthCultNetDaemon owns the local CultNet command-observation boundary: typed command documents enter the CultNet database, the daemon serializes execution through AquaSynthDaemonService, and typed receipt/provider documents come back out. JSON-lines modes are only operator/xeno edges; they must not become second audio/session authorities.
   - Patch scripts lower into explicit graph records before Faust emission.
   - Song-snippet parity is scene-audio pressure, not redistributed fixture data. The worker may derive local tempo/register/scale hints from private clips, but agents must express live rhythm and pitch motion through owned AquaSynth controls until native sequencing sugar has an explicit DSL/lowering owner.
   - Operator graphs own FM topology: operators, modulation edges, carriers, and feedback. Do not hide this authority inside Voice fields.

tests/AquaSynth.Dsl.Tests/AquaSynthDaemonServiceTests.cs

@@ -97,6 +97,52 @@ public sealed class AquaSynthDaemonServiceTests
         Assert.True(File.Exists(Path.Combine(storeRoot, "streams", "stream-sine-stream.cc")));
     }
 
+    [Fact]
+    public async Task DaemonLiveInstrumentSessionProcessesBlocksWithPersistentControlsWhenNativeFaustIsAvailable()
+    {
+        var storeRoot = NewStoreRoot();
+        using var service = new AquaSynthDaemonService(new AquaSynthDaemonOptions(storeRoot));
+
+        var open = await service.OpenInstrumentAsync(new AquaSynthInstrumentOpenCommand(
+            "live-open",
+            "test.live.sine",
+            "test_live_sine",
+            "param path=/macro/gain default=.05 min=0 max=1 step=.001; voice wave=sine freq=440 gain=@/macro/gain attack=.001 sustain=.1 decay=.05",
+            BlockSize: 128,
+            Gain: 1.0f));
+
+        if (!string.Equals(open.Status, "succeeded", StringComparison.Ordinal))
+        {
+            Assert.Contains("Faust", open.FailureMessage, StringComparison.OrdinalIgnoreCase);
+            return;
+        }
+
+        var quiet = await service.ProcessInstrumentBlockAsync(new AquaSynthInstrumentBlockCommand(
+            "live-block-quiet",
+            open.SessionId,
+            FrameCount: 128));
+        var control = await service.ControlInstrumentAsync(new AquaSynthInstrumentControlCommand(
+            "live-control",
+            open.SessionId,
+            new Dictionary<string, float> { ["/macro/gain"] = .3f }));
+        var loud = await service.ProcessInstrumentBlockAsync(new AquaSynthInstrumentBlockCommand(
+            "live-block-loud",
+            open.SessionId,
+            FrameCount: 128));
+
+        Assert.Equal("succeeded", control.Status);
+        Assert.Equal("succeeded", quiet.Status);
+        Assert.Equal("succeeded", loud.Status);
+        Assert.Equal(open.SessionId, quiet.SessionId);
+        Assert.Equal(open.SessionId, loud.SessionId);
+        Assert.Equal((ulong)0, quiet.Sequence);
+        Assert.Equal((ulong)1, loud.Sequence);
+        Assert.True(loud.Rms > quiet.Rms);
+        Assert.True(File.Exists(UriToPath(quiet.Float32Uri)));
+        Assert.True(File.Exists(UriToPath(loud.Float32Uri)));
+        Assert.Equal(128 * sizeof(float), new FileInfo(UriToPath(loud.Float32Uri)).Length);
+    }
+
     [Fact]
     public async Task CultNetDaemonConsumesCommandDocumentsAndPublishesReceiptDocuments()
     {
@@ -123,6 +169,8 @@ public sealed class AquaSynthDaemonServiceTests
             new CultRecordKey("global:aquasynth.cultnet_provider"));
         Assert.NotNull(provider);
         Assert.Contains(AquaSynthDaemonSchemas.InstrumentSampleCommand, provider.CommandSchemas);
+        Assert.Contains(AquaSynthDaemonSchemas.InstrumentOpenCommand, provider.CommandSchemas);
+        Assert.Contains(AquaSynthDaemonSchemas.InstrumentBlockReceipt, provider.ReceiptSchemas);
         Assert.StartsWith(storeRoot, provider.CachePath, StringComparison.OrdinalIgnoreCase);
         Assert.True(Directory.Exists(Path.GetDirectoryName(provider.CachePath)));
     }

tools/AquaSynthDaemon/Program.cs

@@ -134,6 +134,27 @@ internal static class AquaSynthDaemonCli
                     var receipt = await service.StreamAutomationAsync(command).ConfigureAwait(false);
                     await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
                 }
+                else if (string.Equals(envelope.Command, "instrument.open", StringComparison.OrdinalIgnoreCase))
+                {
+                    var command = envelope.Payload.Deserialize<AquaSynthInstrumentOpenCommand>(JsonOptions)
+                        ?? throw new InvalidOperationException("instrument.open payload was empty.");
+                    var receipt = await service.OpenInstrumentAsync(command).ConfigureAwait(false);
+                    await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
+                }
+                else if (string.Equals(envelope.Command, "instrument.control", StringComparison.OrdinalIgnoreCase))
+                {
+                    var command = envelope.Payload.Deserialize<AquaSynthInstrumentControlCommand>(JsonOptions)
+                        ?? throw new InvalidOperationException("instrument.control payload was empty.");
+                    var receipt = await service.ControlInstrumentAsync(command).ConfigureAwait(false);
+                    await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
+                }
+                else if (string.Equals(envelope.Command, "instrument.block", StringComparison.OrdinalIgnoreCase))
+                {
+                    var command = envelope.Payload.Deserialize<AquaSynthInstrumentBlockCommand>(JsonOptions)
+                        ?? throw new InvalidOperationException("instrument.block payload was empty.");
+                    var receipt = await service.ProcessInstrumentBlockAsync(command).ConfigureAwait(false);
+                    await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
+                }
                 else
                 {
                     throw new InvalidOperationException($"Unknown command '{envelope.Command}'.");
@@ -254,6 +275,27 @@ internal static class AquaSynthDaemonCli
                     var receipt = await cultNet.SubmitStreamAsync(command).ConfigureAwait(false);
                     await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
                 }
+                else if (string.Equals(envelope.Command, "instrument.open", StringComparison.OrdinalIgnoreCase))
+                {
+                    var command = envelope.Payload.Deserialize<AquaSynthInstrumentOpenCommand>(JsonOptions)
+                        ?? throw new InvalidOperationException("instrument.open payload was empty.");
+                    var receipt = await cultNet.SubmitOpenAsync(command).ConfigureAwait(false);
+                    await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
+                }
+                else if (string.Equals(envelope.Command, "instrument.control", StringComparison.OrdinalIgnoreCase))
+                {
+                    var command = envelope.Payload.Deserialize<AquaSynthInstrumentControlCommand>(JsonOptions)
+                        ?? throw new InvalidOperationException("instrument.control payload was empty.");
+                    var receipt = await cultNet.SubmitControlAsync(command).ConfigureAwait(false);
+                    await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
+                }
+                else if (string.Equals(envelope.Command, "instrument.block", StringComparison.OrdinalIgnoreCase))
+                {
+                    var command = envelope.Payload.Deserialize<AquaSynthInstrumentBlockCommand>(JsonOptions)
+                        ?? throw new InvalidOperationException("instrument.block payload was empty.");
+                    var receipt = await cultNet.SubmitBlockAsync(command).ConfigureAwait(false);
+                    await output.WriteLineAsync(JsonSerializer.Serialize(receipt, JsonOptions)).ConfigureAwait(false);
+                }
                 else
                 {
                     throw new InvalidOperationException($"Unknown command '{envelope.Command}'.");
@@ -312,6 +354,9 @@ internal static class AquaSynthDaemonCli
             JSON-lines daemon commands:
             {"command":"instrument.sample","payload":{"commandId":"demo","patchId":"demo.patch","faustName":"demo","script":"voice wave=sine freq=440 gain=.2","durationSeconds":0.1}}
             {"command":"instrument.stream","payload":{"commandId":"demo-stream","patchId":"demo.patch","faustName":"demo","script":"param path=/macro/gain default=.2 min=0 max=1 step=.001; voice wave=sine freq=440 gain=@/macro/gain","blockSize":128,"blockCount":4,"controlFrames":[{"block":1,"controls":{"/macro/gain":0.05}}]}}
+            {"command":"instrument.open","payload":{"commandId":"demo-open","patchId":"demo.patch","faustName":"demo","script":"param path=/macro/gain default=.2 min=0 max=1 step=.001; voice wave=sine freq=440 gain=@/macro/gain","blockSize":128}}
+            {"command":"instrument.control","payload":{"commandId":"demo-control","sessionId":"SESSION_FROM_OPEN","controls":{"/macro/gain":0.1}}}
+            {"command":"instrument.block","payload":{"commandId":"demo-block","sessionId":"SESSION_FROM_OPEN","frameCount":128}}
             """).ConfigureAwait(false);
     }