Retain stock coordinator detach after the final snapshot

docs/roadmaps/barc-01-useful-play.md

@@ -533,3 +533,39 @@ Current-UID sampled coverage, SourceLink remote content Unknown and missing
 post-last-snapshot disconnect coverage remain explicit; no fresh live/browser
 eligibility follows. Native seed and requested-unit acceptance remain unknown,
 actor eligibility false and the six useful-play journeys **0/6**.
+
+
+### Owning coordinator detach source qualification (2026-10-05)
+
+The [stock host](../../tests/Broker.NativeProof/LiveHost.fs) now retains the
+owning coordinator's detach even when no later state update arrives. Its existing
+private append-only journal records `kind=detach` with the closed code
+`coordinator-detached`; plugin identifiers and free-form audit reasons are omitted.
+The coordinator's existing generation fence prevents stale stream cleanup from
+closing a replacement session. Non-stock hosts keep their existing audit behavior.
+
+A fresh Release build passed with zero warnings and errors. Eight actual
+[controlled gRPC cases](../../tests/Broker.NativeProof/AttributionControls.fs)
+passed, including detach after the final snapshot without another update and
+stale-generation cleanup after replacement. The 117 portable seed controls include
+refusal of recorded generation invalidation after the terminal snapshot. These
+controlled producers launch no game and grant no actor or command authority.
+
+The original build operation remains `qualified=false`: its final pure proof
+rejected a canonical SDK reference reached through a selected directory alias.
+A separate read-only supplement joined all 307 canonical SDK references to their
+original selected physical pins, verified the exact alias custody, and accepted
+seven actual compiler-input and PE/PDB source joins. It replayed no restore,
+compiler or gRPC control and changed no original result. The accepted compiler
+proof SHA-256 is
+`53969cffe7eee076eb902f1b74067684cbf36630ed7bf0ab737f112630c0dca2`.
+
+This closes the controlled source disconnect coverage gap. Actual native seed
+journal coverage, fresh host/data/settings/mapping custody and a distinct native
+admission remain pending. Prospective programme capacity is eight owned task CLR
+plus two reviewed infrastructure CLR; exact command expectations and refusal of
+unknown processes remain independent. No native copy, seed, browser Arm or
+requested production is admitted by this source qualification.
+**Actor eligibility remains false; requested-unit acceptance is unproved;
+useful play remains 0/6.** SourceLink remote content and telemetry usage remain
+unknown. Publication and installed adoption are separate.

tests/Broker.NativeProof/AttributionControls.fs

@@ -4,6 +4,8 @@ open System
 open System.Collections.Concurrent
 open System.Diagnostics
 open System.Net
+open System.IO
+open System.Text.Json
 open System.Net.Sockets
 open System.Threading
 open Grpc.Net.Client
@@ -53,7 +55,7 @@ module AttributionControls =
         update.Delta <- delta
         update
 
-    let run () = task {
+    let private runCore (journalPath: string option) = task {
         use lifetime = new CancellationTokenSource(TimeSpan.FromSeconds 25.)
         let rows = ConcurrentQueue<struct(Guid * StateUpdate * string)>()
         let subscriptions = ConcurrentBag<IDisposable>()
@@ -77,7 +79,25 @@ module AttributionControls =
         let port = (portSelection.LocalEndpoint :?> IPEndPoint).Port
         portSelection.Stop()
         let audit = ConcurrentQueue<Audit.AuditEvent>()
-        let! host = ServerHost.start { ServerHost.defaultOptions with listenAddress=sprintf "127.0.0.1:%d" port } (Version(1,0)) audit.Enqueue lifetime.Token
+        use detachJournal =
+            match journalPath with
+            | Some path ->
+                require (Path.IsPathFullyQualified path && not(File.Exists path)) "new absolute detach control journal required"
+                new StreamWriter(LiveHost.openStockJournal path)
+            | None -> new StreamWriter(Stream.Null)
+        let detachRows = ConcurrentQueue<string>()
+        let detachGate = obj()
+        let captureDetach kind (value: objnull) = lock detachGate (fun () ->
+            // Exercise the production audit projection and production held-FD writer.
+            let row=JsonSerializer.Serialize {|kind=kind;value=value|}
+            require (System.Text.Encoding.UTF8.GetByteCount(row)+1<=1024 && detachRows.Count<8) "detach control journal bound"
+            detachRows.Enqueue row
+            detachJournal.WriteLine row
+            detachJournal.Flush())
+        let observeAudit event =
+            audit.Enqueue event
+            LiveHost.observeStockAudit true captureDetach event
+        let! host = ServerHost.start { ServerHost.defaultOptions with listenAddress=sprintf "127.0.0.1:%d" port } (Version(1,0)) observeAudit lifetime.Token
         let waitFor name condition = task {
             let stop = DateTimeOffset.UtcNow.AddSeconds 8.
             while not(condition()) do
@@ -141,6 +161,18 @@ module AttributionControls =
 
             // Let the existing owning-generation watchdog detach this controlled stream.
             do! waitFor "old generation detach" (fun () -> BrokerState.session host.Hub |> Option.isNone)
+            if journalPath.IsSome then
+                do! waitFor "detach journal without later snapshot" (fun () -> detachRows.Count=1)
+                require (rows.Count=4) "detach control required an extra state update"
+                use detached=JsonDocument.Parse(detachRows.ToArray().[0])
+                let root=detached.RootElement
+                require (root.GetProperty("kind").GetString()="detach" && root.GetProperty("value").GetProperty("detail").GetString()="coordinator-detached") "bounded detach projection missing"
+                require (root.EnumerateObject() |> Seq.map _.Name |> Set.ofSeq = set["kind";"value"]) "detach projection leaked audit fields"
+                require (root.GetProperty("value").EnumerateObject() |> Seq.map _.Name |> Set.ofSeq = set["detail"]) "detach detail leaked identifiers/reason"
+                let path=journalPath.Value
+                require (File.ReadAllLines(path)=detachRows.ToArray()) "held-FD detach journal not flushed"
+                require (File.GetUnixFileMode(path)=(UnixFileMode.UserRead ||| UnixFileMode.UserWrite)) "detach journal mode changed"
+                printfn "ATTRIBUTION_CONTROL_PASS detach-after-last-snapshot-journaled"
             let! _ = heartbeat "attribution-new"
             use newPush = producer.PushStateAsync(cancellationToken=lifetime.Token)
             do! newPush.RequestStream.WriteAsync(snapshot 1UL 100u)
@@ -155,10 +187,19 @@ module AttributionControls =
             let tick = BrokerState.session host.Hub |> Option.bind(fun session -> (Session.toReading DateTimeOffset.UtcNow session).telemetry) |> Option.map _.tick
             require (tick=Some 100L) "stale stream contaminated replacement"
             printfn "ATTRIBUTION_CONTROL_PASS stale-generation-refused"
-            printfn "ATTRIBUTION_CONTROLS_COMPLETE passed=6 controlledProducer=true gameLaunched=false"
+            if journalPath.IsSome then
+                let beforeDetach=detachRows.Count
+                do! oldPush.RequestStream.CompleteAsync()
+                do! System.Threading.Tasks.Task.Delay(100,lifetime.Token)
+                require (detachRows.Count=beforeDetach && (BrokerState.session host.Hub |> Option.isSome)) "stale stream cleanup emitted detach or closed replacement"
+                printfn "ATTRIBUTION_CONTROL_PASS stale-generation-detach-refused"
+            printfn "ATTRIBUTION_CONTROLS_COMPLETE passed=%d controlledProducer=true gameLaunched=false" (if journalPath.IsSome then 8 else 6)
             return 0
         finally
             lifetime.Cancel()
             (host :> IAsyncDisposable).DisposeAsync().AsTask().GetAwaiter().GetResult()
             for subscription in subscriptions do subscription.Dispose()
     }
+
+    let run () = runCore None
+    let runDetachControls path = runCore (Some path)

tests/Broker.NativeProof/LiveHost.fs

@@ -48,6 +48,14 @@ module LiveHost =
             try File.Delete path with _ -> ()
             reraise()
 
+    // Only the owning-generation detach audit reaches this callback. Retain a
+    // closed refusal code; plugin IDs and free-form reasons stay outside evidence.
+    let observeStockAudit stock (writeKind: string -> objnull -> unit) (event: Audit.AuditEvent) =
+        match event with
+        | Audit.AuditEvent.CoordinatorDetached _ when stock ->
+            writeKind "detach" (box {| detail="coordinator-detached" |})
+        | _ -> ()
+
     let private requiredEnvironment name =
         Environment.GetEnvironmentVariable name
         |> Option.ofObj
@@ -213,7 +221,7 @@ module LiveHost =
                             journal.Flush()
                             stream.Flush(true)
                             journal.Dispose()) }
-        let! host = ServerHost.start { ServerHost.defaultOptions with listenAddress=grpcAddress } (Version(1,0)) ignore lifetime.Token
+        let! host = ServerHost.start { ServerHost.defaultOptions with listenAddress=grpcAddress } (Version(1,0)) (observeStockAudit stock writeKind) lifetime.Token
         try
             let lobby: Lobby.LobbyConfig =
                 { mapName="Avalanche 3.4";gameMode="Skirmish";participants=[{slotIndex=1;kind=ParticipantSlot.ProxyAi;team=0;boundClient=None}];display=Lobby.Headless }

tests/Broker.NativeProof/Program.fs

@@ -232,6 +232,7 @@ module Program =
             (if argv.Length > 0 && argv[0] = "--live-host" then LiveHost.run "barc-live-v1" argv[1..]
              elif argv.Length > 0 && argv[0] = "--tactical-live-host" then LiveHost.run "barc-live-tactical-v1" argv[1..]
              elif argv = [| "--attribution-controls" |] then AttributionControls.run ()
+             elif argv.Length=2 && argv[0]="--attribution-detach-controls" then AttributionControls.runDetachControls argv[1]
              elif argv.Length > 0 && argv[0] = "--stock-tactical-live-host" then LiveHost.run "barc-live-tactical-stock-v1" argv[1..]
              else run listenAddress) |> fun task -> task.GetAwaiter().GetResult()
         with ex ->

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/run-source-controls.py

@@ -17,7 +17,7 @@ def forbidden(*args,**kwargs):raise RuntimeError('portable source controls prohi
 suite=unittest.TestSuite()
 for index,name in enumerate(FILES):
  spec=importlib.util.spec_from_file_location('portable_seed_controls_'+str(index),ROOT/name);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);suite.addTests(unittest.defaultTestLoader.loadTestsFromModule(module))
-expected=115
+expected=117
 if suite.countTestCases()!=expected:raise SystemExit('portable source control roster changed: '+str(suite.countTestCases()))
 result=unittest.TextTestRunner(verbosity=2).run(suite)
 if not result.wasSuccessful()or result.testsRun!=expected or result.skipped:raise SystemExit(1)

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/test_qualify_prearm.py

@@ -85,6 +85,21 @@ def test_reported_host_gap_and_lifetime_refusals(self):
     tactical=[x['value']['tactical']for x in rows if x['kind']=='live-metadata'];tactical[1]['actors'][1]['actor']['lifetime']='99'
    c['inputs']['host']=self.pin(b''.join(json.dumps(x,separators=(',',':')).encode()+b'\n'for x in rows),True)
    with self.assertRaises(q.Refused):self.derive(c)
+ def test_host_invalidation_after_terminal_snapshot_refuses(self):
+  # A terminal snapshot does not erase a later generation failure in the sealed extent.
+  for kind in ['gap','detach','identity-conflict']:
+   with self.subTest(kind=kind):
+    c=self.fixture();rows=[json.loads(x)for x in pathlib.Path(c['inputs']['host']['path']).read_bytes().splitlines()]
+    row=copy.deepcopy(rows[-1]);row.update(sequence=str(len(rows)+1),kind=kind,value={'detail':'controlled terminal failure'})
+    rows.append(row);c['inputs']['host']=self.pin(b''.join(json.dumps(x,separators=(',',':')).encode()+b'\n'for x in rows),True)
+    with self.assertRaisesRegex(q.Refused,'host-generation-refusal'):self.derive(c)
+ def test_stale_generation_failure_after_terminal_snapshot_refuses(self):
+  for detail in ['state sequence gap','coordinator detached','identity changed','process incarnation changed']:
+   with self.subTest(detail=detail):
+    c=self.fixture();rows=[json.loads(x)for x in pathlib.Path(c['inputs']['host']['path']).read_bytes().splitlines()]
+    row=copy.deepcopy(rows[-1]);row.update(sequence=str(len(rows)+1),kind='stale',value={'receivedSequence':'10','detail':detail})
+    rows.append(row);c['inputs']['host']=self.pin(b''.join(json.dumps(x,separators=(',',':')).encode()+b'\n'for x in rows),True)
+    with self.assertRaisesRegex(q.Refused,'host-generation-refusal'):self.derive(c)
  def test_no_fabricated_boolean_contract(self):
   with self.assertRaises(q.Refused):q.derive({'actualPassed':True,'noAutonomousOrders':True},time.monotonic()+1)
   c=self.fixture();pin=c['inputs']['raw'];body=pathlib.Path(pin['path']).read_bytes();pathlib.Path(pin['path']).write_bytes(body+b'{')