mlua-swarm 0.21.0

Swarm engine host built on mlua — long-running stateful runtime with Role/Verb gate, CapToken, 3-stage pipeline, and Middleware overlay.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
//! [`AgentBlockInProcessSpawnerFactory`] — in-process headless LLM
//! agent execution over the `agent-block-core` SDK.
//!
//! ## Design responsibility — a state-less factory
//!
//! The factory is a **kind-level general-purpose builder** — the
//! process-wide infrastructure layer. It does not carry per-agent
//! specialisation (script / `system_prompt` / tools); all agent
//! specialisation belongs to `AgentDef.spec` + `AgentDef.profile`. The
//! old `default_script_path` / `default_project_root` fields were
//! removed — they were the collision source when a single process
//! hosts multiple agent.md files.
//!
//! ## Two modes (via `ScriptSource`)
//!
//! | Mode | Trigger | Chunk |
//! |---|---|---|
//! | **PromptBasedAgent** (default) | `spec.script_path` absent | `ScriptSource::Inline` — the invoker [`build_inline_agent_invoker`] generates, which calls the SDK's `agent` StdPkg module with the resolved `mcp_servers` embedded. NOT the SDK's own `DefaultAgent`: that one passes no tools, so a frontmatter `tools:` line could never reach the model. |
//! | **ScriptBasedAgent** | `spec.script_path = "<path>"` | `ScriptSource::Path(...)` — a caller-provided Lua script, handed to the SDK verbatim. |
//!
//! Neither mode requires the script to agree on an event-kind string: the
//! `host_handler` sink takes any kind as the terminal result (see
//! "Per-task input and result" below for the one reserved exception).
//!
//! `profile.system_prompt` (the agent.md body) is injected into the
//! `_CONTEXT` Lua global through `BlockConfig.context`, and applies to
//! both modes.
//!
//! ## Spec shape (`AgentDef.spec`)
//!
//! This is the settled `spec` contract for `AgentKind::AgentBlock` (GH
//! #86) — every key is optional, and every key the factory reads is
//! listed here:
//!
//! ```jsonc
//! {
//!   "project_root": "<path>",          // optional, default = std::env::current_dir()
//!   "script_path": "<path>",           // optional; absent => PromptBasedAgent mode
//!   "mcp_rpc_timeout_ms": 30000,       // optional, default = 30s
//!   "mcp_servers": [                   // optional; the pool the tool grant selects from
//!     { "name": "outline", "command": "outline-mcp", "args": [] }
//!   ]
//! }
//! ```
//!
//! ## Tool grant (GH #86)
//!
//! The factory reads the effective tool set off `profile.tools` — which
//! is **already** the resolved `Runner::AgentBlockInProcess.tools`
//! whenever the agent declares that Runner, because
//! `compiler::project_bound_agent_for_legacy_factories` overwrites
//! `profile.tools` from the immutable `BoundAgent` snapshot (including
//! with an empty list, so a Blueprint can revoke an agent.md's inherited
//! `tools:` line). No Runner declared → the agent.md line stands. There
//! is deliberately no build hint for this axis: re-deriving the Runner
//! here would bypass the pinned snapshot and let a `Blueprint.runners`
//! edit change an in-flight Run's grant on resume.
//!
//! Enforcement is per mode:
//!
//! | Mode | Enforcement |
//! |---|---|
//! | PromptBasedAgent | Enforced at **server** granularity. Only the `spec.mcp_servers` entries named by an `mcp__<server>__<tool>` entry of the effective set are embedded into the invoker ([`resolve_needed_mcp_servers`]), so the LLM cannot reach an unlisted server — but it CAN reach every tool of a listed one (the SDK exposes a connected server's full tool list). Grant per server, not per tool. |
//! | ScriptBasedAgent | Not enforceable — the script drives its own `mcp.connect`. Declared `mcp__` entries are therefore **rejected at compile time** rather than silently ignored; drop them and let the script own its connections. |
//!
//! Non-`mcp__`-prefixed names (`Read` / `Write` / `WebSearch`) do not
//! select an MCP server and are inert in both modes (see
//! [`mcp_tools_of`]) — the `opts.extra_tools` carry noted on
//! [`resolve_needed_mcp_servers`].
//!
//! ## Per-task input and result (GH #86)
//!
//! Task context reaches this backend through **one seam**:
//! [`WorkerInvocation::context`], the in-process twin of
//! `WorkerPayload.context`, filled once by `InProcSpawner::spawn` from the
//! materialized [`AgentContextView`]. Nothing here peeks at `Ctx`
//! directly, and no `SpawnerAdapter` wrapper re-resolves it — every
//! Lua-visible surface below is derived from that one value:
//!
//! | Lua surface | Source |
//! |---|---|
//! | `_PROMPT` | The step's evaluated `in`, via `inv.prompt` → `BlockConfig.prompt`. A **String** — a structured `in` arrives JSON-stringified, so a script that wants a table calls `std.json.decode(_PROMPT)`. |
//! | `_CONTEXT` | `profile.system_prompt`, via `BlockConfig.context`. |
//! | [`TASK_METADATA_GLOBAL`] (`_TASK_METADATA`) | `view.task_metadata` (the launch's `init_ctx.task_metadata` bag), set through the SDK's `extra_globals` — converted natively, so nesting survives and the chunk text is never rewritten. |
//! | [`AGENT_CTX_GLOBAL`] (`_AGENT_CTX`) | `view.extra` — the Blueprint-declared agent context (`default_agent_ctx` / `AgentMeta.ctx`, GH #21) after `ContextPolicy` filtering. |
//!
//! Both come from [`context_globals`], the single place this mapping is
//! decided; the Lua in-process worker renders the same two globals from
//! it, so a gate is portable between the two backends. `view.steps`
//! (prior-step OUTPUT pointers) is deliberately NOT rendered — see the
//! carrier note in [`crate::core::agent_context`].
//!
//! No server-process env is involved in any of them. The per-task working
//! directory is not a Lua global — it becomes the SDK's `project_root`
//! (see the next section), which surfaces to a script as
//! `std.env.project_root()` and as the default cwd of `sh.exec` and of
//! MCP servers spawned by `mcp.connect`. It does NOT `chdir` the host
//! process, so a bare `io.open("rel/path")` still resolves against the
//! server's own cwd.
//!
//! A script returns its result by calling `bus.emit(<kind>, payload)` —
//! **not** by returning a value from the chunk. Two destinations:
//!
//! | emit kind | Effect |
//! |---|---|
//! | [`ARTIFACT_EVENT_KIND`] (`artifact`) | Stages a named part through [`WorkerInvocation::sink`] (`{name = ..., content = ...}`) and leaves the invocation running. Any number of these. |
//! | anything else | The terminal result, **first emit wins**. [`WorkerResultCaptor`] normalises the payload into [`WorkerResult`]`.value` (`payload.content` → `payload.response` → the whole payload). |
//!
//! Both verdict channels are therefore reachable:
//! `VerdictChannel::Body` compares the terminal value, and
//! `VerdictChannel::Part` compares a staged `"verdict"` part — stage it
//! with `bus.emit("artifact", {name = "verdict", content = "PASS"})`
//! before the terminal emit, and the plain body stays free for the
//! report.
//!
//! ## `project_root` resolution (issue #17, GH #20)
//!
//! `spec.project_root` (above) is only the **compile-time fallback**
//! tier — resolved once in [`AgentBlockInProcessSpawnerFactory::build`],
//! before any `Ctx` exists. Per invocation, [`resolve_project_root`]
//! applies the task-context tier off [`WorkerInvocation::context`] (GH
//! #20 Contract C — see [`crate::core::agent_context`] for the full
//! narrative) with this priority (highest first):
//!
//! 1. `view.work_dir` — Task-level, set by `TaskInputMiddleware` from
//!    the launch's `init_ctx.work_dir`.
//! 2. `view.project_root` — same middleware, `init_ctx.project_root`.
//! 3. `spec.project_root` / `std::env::current_dir()` (the compile-time
//!    fallback baked into [`AgentBlockSettings`] above).
//!
//! This lets a single Blueprint's `AgentDef.spec.project_root` (fixed at
//! compile time) be overridden per task launch, so the same Blueprint
//! can run against different caller-supplied project roots without a
//! `spec` edit.
//!
//! ## SDK paths introduced from v0.22.0 through v0.27.0
//!
//! | Version | Feature | Use case |
//! |---|---|---|
//! | v0.22.0 | `bus.emit(kind, payload, id?)` Lua bridge | script → host event push |
//! | v0.23.0 | `BlockConfig.host_handlers` | Pre-install a Rust handler on the EventBus |
//! | v0.24.0 | `BlockConfig.auto_serve_bus` | SDK embed drives the dispatcher in the background |
//! | v0.25.0 | `BlockConfig.shutdown_token` + `BlockError::Cancelled` + `Send` on `run()` | `tokio::spawn` and external cancel |
//! | v0.26.0 | `ScriptSource` / `PromptSource` / `SecretKeySource` enums plus the embedded `DefaultAgent` invoker (breaking) | Script becomes optional at the SDK level |
//! | v0.27.0 | Embed the `compile_loop` StdPkg into core | `require("compile_loop")` hits directly |

use crate::core::agent_context::AgentContextView;
use crate::worker::adapter::{InProcSpawner, WorkerError, WorkerInvocation, WorkerResult};
use agent_block_core::bus::dispatcher::Handler;
use agent_block_core::host::{PromptSource, ScriptSource};
use agent_block_core::{run, BlockConfig};
use agent_block_types::error::BlockError;
use async_trait::async_trait;
use serde_json::Value;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::oneshot;

/// Host-side handler that fires when the Lua script (or the
/// DefaultAgent invoker) calls `bus.emit(<kind>, payload)`. It folds
/// the payload into a [`WorkerResult`] and forwards it on the
/// [`oneshot::Sender`].
///
/// This is **an AgentBlock-internal helper**. Different SDK paths use
/// different event names and payload shapes — the DefaultAgent
/// invoker's `agent_result` event carries the entire `agent.run`
/// return value (`{content, messages, num_turns, ok, usage}`), while a
/// caller script's `worker_result` event carries `{ok, response}`. The
/// captor keeps those quirks contained and **normalises them**, so
/// callers (flow.ir, the engine, higher-level Workers) always see the
/// same single form: "the raw LLM response is `WorkerResult.value`".
///
/// Value extraction priority (the normalisation policy that hides the
/// SDK quirks):
///
/// 1. `payload.content` — from the DefaultAgent invoker / `agent.run`
///    return value; carried as a string.
/// 2. `payload.response` — the caller script's `worker_result`
///    convention; free-form.
/// 3. Fallback: the whole payload — for custom shapes that carry
///    neither of the above.
///
/// `ok` extraction: `payload.ok` if present, otherwise `true` — the
/// DefaultAgent invoker includes `ok`, so this recovers it.
///
/// This is the core of the observation #2 fix. The previous
/// implementation did not consult (1); it only fell back
/// `(2) → (3)`. On the DefaultAgent path that pushed the whole
/// `agent_result` object into `WorkerResult.value`, which then rode
/// through the chain and hit the next step's prompt via
/// JSON-stringification — burning 50-60% of the tokens on
/// boilerplate. Pulling out (1) first normalises the chain to a single
/// LLM raw-text carry and brings the Worker pattern up to the token
/// efficiency of the Phase 3 WS Operator path.
///
/// # Named parts (GH #86)
///
/// One event kind is reserved: [`ARTIFACT_EVENT_KIND`]. An emit under that
/// kind is staged as an [`OutputEvent::Artifact`] through
/// [`WorkerInvocation::sink`] and does **not** complete the invocation, so
/// a script may stage any number of named parts and then finish with its
/// terminal emit as usual. This is what makes `VerdictChannel::Part`
/// reachable from this backend — the engine's completion-time contract
/// check looks for a staged `"verdict"` artifact.
///
/// Reserving a kind is a deliberate step back from "kind-agnostic": with
/// two destinations there is now something to route, so the script side
/// has one string to coordinate. Every other kind keeps the old
/// first-emit-wins result behavior.
struct WorkerResultCaptor {
    tx: Mutex<Option<oneshot::Sender<WorkerResult>>>,
    /// Intake for staged named parts; `None` when the caller path did not
    /// wire one (an `artifact` emit then degrades to a `tracing::warn!`
    /// rather than silently vanishing).
    sink: Option<Arc<dyn crate::worker::output::OutputSink>>,
    /// The agent's Blueprint-declared model
    /// ([`AgentBlockSettings::declared_model`]) — the fallback for
    /// `stats.model` when the payload does not report one at runtime.
    declared_model: Option<String>,
}

impl WorkerResultCaptor {
    /// SDK-quirks normalisation: extract `(value, ok)` from a
    /// `bus.emit` payload. `pub(crate)` so both callers and unit tests
    /// can reach it.
    fn extract(payload: &Value) -> (Value, bool) {
        let ok = payload.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
        let value = payload
            .get("content")
            .cloned()
            .or_else(|| payload.get("response").cloned())
            .unwrap_or_else(|| payload.clone());
        (value, ok)
    }

    /// Stats-sidecar extraction (per-step run stats): the DefaultAgent
    /// invoker's `agent_result` payload carries the full `agent.run`
    /// return, whose `usage` (`{input_tokens, output_tokens,
    /// total_tokens}`, all turns summed) and `num_turns` used to be
    /// DROPPED here — the exact gap this recovers. `None` when the
    /// payload carries none of them (caller-script `worker_result`
    /// shapes). The raw `usage` object also rides as `adapter_data` so
    /// provider-specific detail (cache tokens etc.) survives.
    ///
    /// A payload-level `"model"` string is the **runtime-observed** model
    /// and is adopted verbatim; it outranks the Blueprint-declared
    /// fallback applied in [`Handler::call`], because what actually
    /// served the attempt beats what was asked for.
    fn extract_stats(payload: &Value) -> Option<crate::store::trace::WorkerStats> {
        let usage_raw = payload.get("usage");
        let usage = usage_raw.and_then(|u| {
            let input = u.get("input_tokens").and_then(|v| v.as_u64());
            let output = u.get("output_tokens").and_then(|v| v.as_u64());
            match (input, output) {
                (Some(i), Some(o)) => Some(crate::store::trace::TokenUsage {
                    input_tokens: i,
                    output_tokens: o,
                    total_tokens: u
                        .get("total_tokens")
                        .and_then(|v| v.as_u64())
                        .unwrap_or(i + o),
                }),
                _ => None,
            }
        });
        let num_turns = payload
            .get("num_turns")
            .and_then(|v| v.as_u64())
            .map(|n| n as u32);
        let model = payload
            .get("model")
            .and_then(|v| v.as_str())
            .map(str::to_string);
        if usage.is_none() && num_turns.is_none() && model.is_none() {
            return None;
        }
        Some(crate::store::trace::WorkerStats {
            worker_kind: Some("agent_block".to_string()),
            model,
            usage,
            num_turns,
            adapter_data: usage_raw.cloned(),
        })
    }

    /// GH #86: stage one named part from an [`ARTIFACT_EVENT_KIND`] emit.
    ///
    /// `payload.name` (string) is required — an emit without it cannot
    /// address a part, so it is reported to the script as an error rather
    /// than dropped. `payload.content` is the body; absent means the part
    /// is staged empty (`Value::Null`), which is still a distinct,
    /// addressable staging event.
    async fn stage_artifact(&self, payload: &Value) -> Result<(), BlockError> {
        let name = payload
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or_else(|| {
                BlockError::Runtime(format!(
                    "bus.emit(\"{ARTIFACT_EVENT_KIND}\", ...) requires a string `name` field \
                     naming the part (got: {payload})"
                ))
            })?
            .to_string();
        let Some(sink) = self.sink.as_ref() else {
            tracing::warn!(
                artifact = %name,
                "agent-block staged an artifact but no OutputSink is wired for this \
                 invocation; the part is dropped"
            );
            return Ok(());
        };
        let content = payload.get("content").cloned().unwrap_or(Value::Null);
        sink.emit(crate::worker::output::OutputEvent::Artifact {
            name: name.clone(),
            content: crate::worker::output::ContentRef::Inline { value: content },
        })
        .await
        .map_err(|e| BlockError::Runtime(format!("staging artifact '{name}': {e}")))?;
        Ok(())
    }
}

#[async_trait]
impl Handler for WorkerResultCaptor {
    async fn call(
        &self,
        kind: String,
        _id: String,
        payload: Value,
        _meta: Value,
    ) -> Result<Value, BlockError> {
        // GH #86: the one reserved kind routes to the named-part intake and
        // leaves the invocation running; everything else is the terminal
        // result (first emit wins).
        if kind == ARTIFACT_EVENT_KIND {
            self.stage_artifact(&payload).await?;
            return Ok(Value::Null);
        }
        let (value, ok) = Self::extract(&payload);
        let stats = Self::extract_stats(&payload);
        // Even when the SDK payload carries no usage (script-side
        // `worker_result` shapes), the boundary still knows its own
        // kind — surface it so `StepEntry.worker_kind` is never empty.
        let mut wr = WorkerResult { value, ok, stats }.ensure_worker_kind("agent_block");
        // `agent.run` returns no model, so without this the sidecar's
        // `model` was always empty on this backend. Fall back to the
        // Blueprint-declared `profile.model`, mirroring the subprocess
        // backend's baked `{model}`. Precedence: runtime-observed
        // (`extract_stats`) > declared > none — `get_or_insert_with`
        // encodes exactly that (it is a no-op once a value is present).
        if let (Some(declared), Some(stats)) = (self.declared_model.as_deref(), wr.stats.as_mut()) {
            stats.model.get_or_insert_with(|| declared.to_string());
        }
        if let Ok(mut guard) = self.tx.lock() {
            if let Some(tx) = guard.take() {
                let _ = tx.send(wr);
            }
        }
        Ok(Value::Null)
    }
}

/// The Lua global carrying the launch's `init_ctx.task_metadata` bag into
/// a script, set through the SDK's `extra_globals` (agent-block-core
/// v0.30+).
///
/// Underscore-prefixed to sit alongside the SDK's own globals.
/// `_PROMPT` / `_CONTEXT` / `_SCRIPT_NAME` are reserved by the SDK and
/// must not be used here; this name is ours.
pub const TASK_METADATA_GLOBAL: &str = "_TASK_METADATA";

/// The Lua global carrying the Blueprint-declared agent context — the
/// `AgentContextView.extra` bag, which `AgentContextMiddleware` fills from
/// `Blueprint.default_agent_ctx` / `AgentMeta.ctx` (GH #21) after applying
/// `ContextPolicy`.
///
/// The WS Operator lane has always received these as `{key}: {value}`
/// lines of the Spawn directive header
/// (`AgentContextView::to_directive_header`); this is the in-process
/// equivalent. Delivered as ONE table rather than one global per key,
/// because the keys are Blueprint-author-chosen and must not be able to
/// shadow `_PROMPT` / `_CONTEXT` / `_SCRIPT_NAME` or each other's
/// namespace.
pub const AGENT_CTX_GLOBAL: &str = "_AGENT_CTX";

/// The Lua globals this backend derives from the materialized context
/// view — the single place the mapping "view field → Lua global" is
/// decided, so [`crate::blueprint::compiler`]'s Lua worker can render the
/// same surface and keep a gate portable between the two in-process
/// backends.
///
/// Absent / empty inputs contribute no entry at all (rather than an empty
/// table), so a script sees `nil` and can branch on presence — the same
/// "insert nothing when absent" contract the rest of this axis follows.
pub fn context_globals(view: Option<&AgentContextView>) -> HashMap<String, Value> {
    let mut globals = HashMap::new();
    let Some(view) = view else {
        return globals;
    };
    if let Some(meta) = view.task_metadata.clone() {
        globals.insert(TASK_METADATA_GLOBAL.to_string(), meta);
    }
    if !view.extra.is_empty() {
        globals.insert(
            AGENT_CTX_GLOBAL.to_string(),
            Value::Object(view.extra.clone()),
        );
    }
    globals
}

/// The one `bus.emit` kind this backend reserves: an emit under it stages
/// a named part instead of completing the invocation (GH #86).
///
/// Payload shape: `{ name = "<part>", content = <any> }`. `name` is
/// required. Reaching `VerdictChannel::Part` from a script means staging
/// `{ name = "verdict", content = "PASS" }` before the terminal emit.
pub const ARTIFACT_EVENT_KIND: &str = "artifact";

/// Settings baked per `AgentDef` — the static portion of one
/// invocation. Everything task-dependent (`project_root` /
/// `task_metadata`) is resolved per invocation off
/// [`WorkerInvocation::context`] instead, so this struct is built once at
/// compile time and shared by every dispatch of the agent.
///
/// v0.28.0 adopted `BlockConfig.host_handler` (a kind-agnostic
/// single sink backed by `EventBus::on_any`); the older
/// `result_event_kind: String` field (which required the caller /
/// script to coordinate a kind string) is gone. One captor per
/// invocation is enough, so a single sink is enough.
#[derive(Clone)]
struct AgentBlockSettings {
    /// The chunk to run — see [`ScriptPlan`].
    script: ScriptSource,
    /// Compile-time fallback cwd: `spec.project_root`, else
    /// `env::current_dir()`. Outranked per invocation by the context
    /// view's `work_dir` / `project_root`.
    spec_project_root: PathBuf,
    mcp_rpc_timeout: Duration,
    /// Agent persona — the `system_prompt` composed from the agent.md
    /// body and frontmatter. `None` maps to `BlockConfig.context = None`
    /// for backwards compatibility with the old path.
    profile_context: Option<String>,
    /// The agent.md frontmatter `model:` line (`profile.model`), baked at
    /// compile time. Observational only — this backend does not select a
    /// model with it (the SDK owns that); it is the declared fallback for
    /// the per-step `stats.model` sidecar, so a step of an agent whose BP
    /// names a model is attributable even though `agent.run` reports
    /// none. Sibling of the subprocess backend's baked `{model}`.
    declared_model: Option<String>,
}

/// One invocation's worth of an `agent-block-core` SDK call — the
/// `WorkerFn` body.
///
/// Registers the result captor through the v0.28.0 `host_handler`
/// (single, kind-agnostic fallback). The plural `host_handlers`
/// (string-keyed routing) is not needed — one captor per invocation is
/// enough, and there is no script-side event-kind string to coordinate.
async fn run_agent_block_worker(
    settings: Arc<AgentBlockSettings>,
    inv: WorkerInvocation,
) -> Result<WorkerResult, WorkerError> {
    let (tx, rx) = oneshot::channel();
    let captor: Arc<dyn Handler> = Arc::new(WorkerResultCaptor {
        tx: Mutex::new(Some(tx)),
        sink: inv.sink.clone(),
        declared_model: settings.declared_model.clone(),
    });

    // GH #86: the task-context tier, read off the ONE in-process seam
    // (`WorkerInvocation.context`, filled by `InProcSpawner::spawn` from
    // the materialized `AgentContextView`) instead of a hand-rolled `Ctx`
    // peek in a spawner wrapper.
    let project_root = resolve_project_root(inv.context.as_ref(), &settings.spec_project_root);

    // Bridge the shutdown token: forward `WorkerInvocation.cancel_token`
    // into the SDK's `shutdown_token` if one is set; otherwise use a
    // fresh token (no external cancel).
    let shutdown_token = inv.cancel_token.clone().unwrap_or_default();

    let mut builder = BlockConfig::builder(settings.script.clone(), project_root)
        .mcp_rpc_timeout(settings.mcp_rpc_timeout)
        .prompt(PromptSource::Inline(inv.prompt))
        .host_handler(captor)
        .auto_serve_bus(true)
        .shutdown_token(shutdown_token.clone());
    if let Some(system) = settings.profile_context.clone() {
        builder = builder.context(PromptSource::Inline(system));
    }
    // The task-context globals. `extra_globals` (SDK v0.30) sets each entry
    // on both Isles before the chunk runs, converting the JSON natively —
    // so nothing is spliced into the script text: no line-number shift, no
    // `script_dir` / `package.path` disturbance, and no string-escaping
    // trust boundary around caller-supplied values.
    let globals = context_globals(inv.context.as_ref());
    if !globals.is_empty() {
        builder = builder.extra_globals(globals);
    }
    let config = builder.build();

    let run_handle = tokio::spawn(run(config));
    let run_result = run_handle
        .await
        .map_err(|e| WorkerError::Failed(format!("agent-block task join: {e}")))?;
    run_result.map_err(|e| WorkerError::Failed(format!("agent-block run failed: {e}")))?;

    rx.await.map_err(|_| {
        WorkerError::Failed("agent-block script finished without emitting result via bus".into())
    })
}

// ─── tools / mcp_servers resolution ───────────────────────────────────────

/// Cross-reference the agent's declared tool set (see
/// [`resolve_effective_tools`] for which tier that comes from) with
/// `spec.mcp_servers` (the `"server name" → command + args` mapping
/// provided by the `AgentDef` literal cascade) and resolve the
/// `mcp_servers` config actually exposed to the LLM for this invocation.
///
/// Algorithm:
///
/// 1. Extract `mcp__<server>__<tool>` patterns from `declared_tools`;
///    collect the `<server>` names.
/// 2. Filter `spec.mcp_servers` to just the entries whose name is in
///    that set.
///
/// This is the response to observation #3 — do not hand the LLM
/// `mcp_servers` it does not need (only the servers the declaration
/// explicitly asks for), and equally do not expose servers the
/// declaration does not know about even if the spec carries them
/// (caller intent wins).
///
/// CC built-in tools (non-`mcp__`-prefixed names like `Read` / `Write`
/// / `WebSearch`) are out of scope here; handling those lives in a
/// different layer — a carry that would come through a future
/// `opts.extra_tools` Rust implementation.
pub fn resolve_needed_mcp_servers(
    declared_tools: &[String],
    spec_mcp_servers: &[Value],
) -> Vec<Value> {
    use std::collections::HashSet;
    // Step 1: server names from `mcp__<server>__<tool>` patterns in the
    // declared tool set.
    let needed: HashSet<&str> = declared_tools
        .iter()
        .filter_map(|t| {
            let rest = t.strip_prefix("mcp__")?;
            // Split `<server>__<tool>` at the first `__`.
            let idx = rest.find("__")?;
            Some(&rest[..idx])
        })
        .collect();

    // Step 2: filter `spec.mcp_servers` down to entries whose name is
    // in `needed`.
    spec_mcp_servers
        .iter()
        .filter(|cfg| {
            cfg.get("name")
                .and_then(|n| n.as_str())
                .map(|name| needed.contains(name))
                .unwrap_or(false)
        })
        .cloned()
        .collect()
}

/// GH #86 — the subset of an effective tool set that names an MCP server,
/// i.e. the only entries this backend's grant model can act on.
///
/// Everything else (`Read` / `Write` / `WebSearch` …) selects no server and
/// is inert here, so it is neither embedded nor treated as a grant that
/// must be honored — see [`resolve_needed_mcp_servers`]'s `opts.extra_tools`
/// carry. Used by the ScriptBasedAgent guard in
/// [`AgentBlockInProcessSpawnerFactory::build`], which must not fail an
/// agent whose declared tools are all inert.
fn mcp_tools_of(tools: &[String]) -> Vec<&str> {
    tools
        .iter()
        .filter(|t| t.starts_with("mcp__"))
        .map(String::as_str)
        .collect()
}

/// Build the inline Lua script used on the PromptBasedAgent path (when
/// `spec.script_path` is absent). Instead of the SDK's embedded
/// `DEFAULT_AGENT_INVOKER` (which passes no tools), this embeds
/// `mcp_servers` as a Lua literal table and hands it to `agent.run`.
///
/// This is the core of the observation #3 fix. The old DefaultAgent
/// path had no way to deliver a frontmatter `tools:` line to the SDK.
/// This inline path bakes the `profile.tools` → `mcp_servers` config
/// into the Lua source, so the LLM can actually make tool calls.
///
/// The JSON-stringify + `std.json.decode` route was ruled out because
/// the SDK environment cannot `require` the `std` module (no
/// `package.preload['std']` field), so we take the JSON → Lua-literal
/// conversion on the Rust side and embed the result directly. The
/// event name is `agent_result` — the same convention the SDK's
/// internal `DEFAULT_AGENT_INVOKER` uses.
pub fn build_inline_agent_invoker(mcp_servers: &[Value]) -> ScriptSource {
    let mcp_lua = json_array_to_lua_literal(mcp_servers);
    let source = format!(
        r##"local agent = require("agent")
local mcp_servers = {mcp_lua}
local r = agent.run({{
    prompt = _PROMPT,
    system = _CONTEXT,
    mcp_servers = mcp_servers,
}})
bus.emit("agent_result", r)
"##
    );
    ScriptSource::Inline {
        source,
        name: "mlua_swarm_engine_default_agent_invoker.lua".into(),
    }
}

/// Convert a JSON `Value` into a Lua literal expression, for embedding
/// into the inline script. Lua string escaping is delegated to Rust's
/// `{:?}` `Debug` output — Lua syntax is compatible with the escapes
/// it produces (`"`, `\\`, `\n`, `\r`, `\t`, and so on). Edge cases
/// like `\0` or unusual Unicode escapes are outside the scope of this
/// use.
fn json_to_lua_literal(v: &Value) -> String {
    match v {
        Value::Null => "nil".to_string(),
        Value::Bool(b) => b.to_string(),
        Value::Number(n) => n.to_string(),
        Value::String(s) => format!("{s:?}"),
        Value::Array(arr) => {
            let items: Vec<String> = arr.iter().map(json_to_lua_literal).collect();
            format!("{{{}}}", items.join(", "))
        }
        Value::Object(map) => {
            let items: Vec<String> = map
                .iter()
                .map(|(k, v)| format!("[{k:?}]={}", json_to_lua_literal(v)))
                .collect();
            format!("{{{}}}", items.join(", "))
        }
    }
}

/// Convert a `Vec<Value>` into a Lua literal sequence. An empty array
/// becomes `{}` — a Lua empty table.
fn json_array_to_lua_literal(arr: &[Value]) -> String {
    if arr.is_empty() {
        return "{}".to_string();
    }
    let items: Vec<String> = arr.iter().map(json_to_lua_literal).collect();
    format!("{{{}}}", items.join(", "))
}

// ─── SpawnerFactory ───────────────────────────────────────────────────────

/// The compile-time (`spec` / `env::current_dir()`) fallback tier of the
/// `project_root` priority chain (issue #17) — the tail two links of
/// **`ctx.meta.runtime` `work_dir` > `ctx.meta.runtime` `project_root` >
/// `spec.project_root` > `env::current_dir()`**. Extracted as a standalone
/// pure fn so it is independently testable without needing a full `Ctx` /
/// `SpawnerAdapter` round-trip.
fn resolve_spec_project_root(spec: &Value) -> PathBuf {
    match spec.get("project_root").and_then(|v| v.as_str()) {
        Some(s) => PathBuf::from(s),
        None => std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")),
    }
}

/// Apply the task-context tier on top of the compile-time fallback:
/// **`view.work_dir` > `view.project_root` > `spec_fallback`**.
///
/// `work_dir` outranks `project_root` because it names the exact directory
/// this specific worker should run from. A `None` view (no `Ctx` on the
/// caller path) leaves the compile-time fallback in place.
fn resolve_project_root(view: Option<&AgentContextView>, spec_fallback: &Path) -> PathBuf {
    view.and_then(|v| v.work_dir.as_deref().or(v.project_root.as_deref()))
        .map(PathBuf::from)
        .unwrap_or_else(|| spec_fallback.to_path_buf())
}

/// The `SpawnerFactory` for AgentBlock. `KIND = AgentKind::AgentBlock`.
///
/// **State-less.** One factory per process; every `AgentDef` uses it
/// as a shared builder. Per-agent specialisation stays **entirely
/// inside `AgentDef.spec` + `AgentDef.profile`** — the old
/// `default_script_path` / `default_project_root` fields are gone.
///
/// Naming convention: `<WorkerIMPL><AdapterType>SpawnerFactory` — an
/// AgentBlock worker on the InProcess adapter.
pub struct AgentBlockInProcessSpawnerFactory;

impl Default for AgentBlockInProcessSpawnerFactory {
    fn default() -> Self {
        Self
    }
}

impl AgentBlockInProcessSpawnerFactory {
    /// Stateless constructor — equivalent to `Default::default()`.
    pub fn new() -> Self {
        Self
    }
}

impl crate::blueprint::compiler::SpawnerFactoryKind for AgentBlockInProcessSpawnerFactory {
    const KIND: crate::blueprint::AgentKind = crate::blueprint::AgentKind::AgentBlock;
    type Worker = AgentBlockWorker;
}

impl crate::blueprint::compiler::SpawnerFactory for AgentBlockInProcessSpawnerFactory {
    fn build(
        &self,
        agent_def: &crate::blueprint::AgentDef,
        _hint: Option<&Value>,
    ) -> Result<
        Arc<dyn crate::worker::adapter::SpawnerAdapter>,
        crate::blueprint::compiler::CompileError,
    > {
        let agent_name = agent_def.name.clone();
        let spec = &agent_def.spec;

        // Resolve the actual mcp_servers config to pass to the real LLM by
        // combining the effective tool set with spec.mcp_servers (the first
        // axis of AgentDef literal cascade — a "server name → command +
        // args" mapping). The result is JSON-embedded into the Lua source by
        // build_inline_agent_invoker and flows into
        // `agent.run({mcp_servers=...})`.
        //
        // `profile.tools` IS the effective set: when the agent declares a
        // `Runner::AgentBlockInProcess`, the compiler has already overwritten
        // `profile.tools` with that Runner's `tools` off the pinned
        // `BoundAgent` snapshot (`project_bound_agent_for_legacy_factories`),
        // including with an empty list. No Runner declared → the agent.md
        // `tools:` line stands as-is. See the module doc's "Tool grant".
        let effective_tools: Vec<String> = agent_def
            .profile
            .as_ref()
            .map(|p| p.tools.clone())
            .unwrap_or_default();
        let spec_mcp_servers: Vec<Value> = spec
            .get("mcp_servers")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        let needed_mcp_servers = resolve_needed_mcp_servers(&effective_tools, &spec_mcp_servers);

        // script: `spec.script_path` absent → PromptBasedAgent (the new Inline
        //         path, embedding tools and calling agent.run); present →
        //         ScriptBasedAgent (a caller-provided script path where tools
        //         are the caller's responsibility). Event-kind string
        //         dependency was retired — the `host_handler` single sink
        //         captures every kind.
        let script = match spec.get("script_path").and_then(|v| v.as_str()) {
            Some(s) => {
                // GH #86: a caller script drives its own `mcp.connect`, so the
                // host has no choke point to enforce an MCP grant through —
                // the Inline invoker's "embed exactly the declared servers"
                // lever does not exist on this path. Declaring MCP tools here
                // would be a promise the runtime cannot keep, so it is
                // rejected at compile time instead of silently ignored.
                //
                // Only `mcp__`-prefixed entries trigger this: everything else
                // selects no server and is inert in both modes, so an agent.md
                // `tools: Read, WebSearch` line must not fail a script-mode
                // agent that compiled before this guard existed.
                let mcp_tools = mcp_tools_of(&effective_tools);
                if !mcp_tools.is_empty() {
                    return Err(crate::blueprint::compiler::CompileError::InvalidSpec {
                        name: agent_name,
                        msg: format!(
                            "agent_block ScriptBasedAgent mode (spec.script_path = {s:?}) cannot \
                             enforce an MCP tool grant: the script opens its own connections via \
                             `mcp.connect`, so the declared tools ({}) would be unenforceable. \
                             Either drop spec.script_path to use PromptBasedAgent mode (where the \
                             declared servers ARE the only ones embedded into the invoker), or \
                             drop the mcp__ entries and let the script own its connections.",
                            mcp_tools.join(", ")
                        ),
                    });
                }
                ScriptSource::Path(PathBuf::from(s))
            }
            None => build_inline_agent_invoker(&needed_mcp_servers),
        };

        // issue #17: this is the compile-time fallback tier only —
        // `spec.project_root`, then `env::current_dir()`. No `Ctx` exists
        // yet at `build()` time, so the higher-priority task-context tier
        // cannot be consulted here; `run_agent_block_worker` applies it per
        // invocation off `WorkerInvocation.context` (see the module-level
        // "`project_root` resolution" doc).
        let spec_project_root = resolve_spec_project_root(spec);
        let mcp_rpc_timeout = match spec.get("mcp_rpc_timeout_ms").and_then(|v| v.as_u64()) {
            Some(ms) => Duration::from_millis(ms),
            None => Duration::from_secs(30),
        };
        let profile_context = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
        // Same source the subprocess backend bakes its `{model}`
        // placeholder from; here it only ever reaches the per-step stats
        // sidecar (see `AgentBlockSettings::declared_model`).
        let declared_model = agent_def.profile.as_ref().and_then(|p| p.model.clone());

        let settings = Arc::new(AgentBlockSettings {
            script,
            spec_project_root,
            mcp_rpc_timeout,
            profile_context,
            declared_model,
        });

        // A plain `InProcSpawner` with this agent's single route. GH #86
        // removed the `AgentBlockCtxAwareSpawner` wrapper that used to sit
        // here purely to re-resolve `ctx.meta.runtime` at spawn time: the
        // task-context tier now arrives on `WorkerInvocation.context`, the
        // same seam every other in-process worker reads, so the worker fn
        // resolves it itself and no bespoke adapter is needed.
        let worker_fn: crate::worker::adapter::WorkerFn = Arc::new(move |inv| {
            let settings = settings.clone();
            Box::pin(run_agent_block_worker(settings, inv))
        });
        let mut sp: InProcSpawner<AgentBlockWorker> = InProcSpawner::<AgentBlockWorker>::typed();
        sp.registry.insert(agent_name, worker_fn);
        Ok(Arc::new(sp))
    }
}

/// Concrete Worker type for the AgentBlock kind — the handle for an
/// LLM call routed through the `agent-block-core` SDK. Embeds a
/// `WorkerJoinHandler` to carry the async signal. The intent is to
/// eventually keep the SDK-specific quirks — the `agent_result` event
/// name, payload shape, shutdown-token bridging, agent_result.content
/// normalisation — contained inside this struct. Today it lands as a
/// thin shape holding only the async signal; Phase B adds the
/// normalisation layer here and structurally eliminates the
/// token-boilerplate waste observed in observation #2.
pub struct AgentBlockWorker {
    /// The completion-signal handle for this agent-block SDK call's
    /// spawned task.
    pub handler: crate::worker::WorkerJoinHandler,
}

impl From<crate::worker::WorkerJoinHandler> for AgentBlockWorker {
    fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
        Self { handler }
    }
}

#[async_trait]
impl crate::worker::Worker for AgentBlockWorker {
    fn id(&self) -> &crate::types::WorkerId {
        &self.handler.worker_id
    }
    fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
        self.handler.cancel.clone()
    }
    async fn join(self: Box<Self>) -> Result<(), WorkerError> {
        self.handler.await_completion().await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::agent_context::{TASK_METADATA_KEY, TASK_PROJECT_ROOT_KEY, TASK_WORK_DIR_KEY};

    #[test]
    fn resolve_needed_mcp_servers_filters_by_tool_prefix() {
        let tools = vec![
            "mcp__semantic-scholar__search_papers".to_string(),
            "mcp__semantic-scholar__get_paper".to_string(),
            "Read".to_string(),
            "mcp__outline__list_docs".to_string(),
            "WebSearch".to_string(),
        ];
        let spec_servers = vec![
            serde_json::json!({"name": "semantic-scholar", "command": "ss-mcp", "args": []}),
            serde_json::json!({"name": "outline", "command": "outline-mcp", "args": []}),
            serde_json::json!({"name": "unused", "command": "nope", "args": []}),
        ];
        let needed = resolve_needed_mcp_servers(&tools, &spec_servers);
        assert_eq!(needed.len(), 2, "got: {needed:?}");
        let names: Vec<&str> = needed
            .iter()
            .filter_map(|c| c.get("name").and_then(|n| n.as_str()))
            .collect();
        assert!(names.contains(&"semantic-scholar"));
        assert!(names.contains(&"outline"));
        assert!(!names.contains(&"unused"), "unused server is filtered out");
    }

    #[test]
    fn resolve_needed_mcp_servers_returns_empty_when_no_mcp_tools() {
        let tools = vec!["Read".to_string(), "WebSearch".to_string()];
        let spec_servers =
            vec![serde_json::json!({"name": "outline", "command": "outline-mcp", "args": []})];
        let needed = resolve_needed_mcp_servers(&tools, &spec_servers);
        assert!(
            needed.is_empty(),
            "no mcp__-prefixed tools → empty result, got: {needed:?}"
        );
    }

    #[test]
    fn build_inline_agent_invoker_embeds_mcp_servers_as_lua_literal() {
        let servers =
            vec![serde_json::json!({"name": "outline", "command": "outline-mcp", "args": []})];
        let script = build_inline_agent_invoker(&servers);
        match script {
            ScriptSource::Inline { source, name } => {
                assert!(name.ends_with(".lua"));
                assert!(source.contains("require(\"agent\")"));
                assert!(source.contains("mcp_servers = mcp_servers"));
                assert!(source.contains("bus.emit(\"agent_result\""));
                // Lua literal embed (= keys [\"name\"]=\"outline\" form)
                assert!(source.contains("[\"name\"]=\"outline\""));
                assert!(source.contains("[\"command\"]=\"outline-mcp\""));
                assert!(source.contains("[\"args\"]={}"), "args empty array literal");
            }
            other => panic!("expected Inline, got: {other:?}"),
        }
    }

    #[test]
    fn build_inline_agent_invoker_with_empty_servers_still_valid() {
        let script = build_inline_agent_invoker(&[]);
        match script {
            ScriptSource::Inline { source, .. } => {
                assert!(source.contains("local mcp_servers = {}"));
            }
            other => panic!("expected Inline, got: {other:?}"),
        }
    }

    #[test]
    fn json_to_lua_literal_handles_primitives_and_nested() {
        assert_eq!(json_to_lua_literal(&serde_json::json!(null)), "nil");
        assert_eq!(json_to_lua_literal(&serde_json::json!(true)), "true");
        assert_eq!(json_to_lua_literal(&serde_json::json!(42)), "42");
        assert_eq!(json_to_lua_literal(&serde_json::json!("hi")), "\"hi\"");
        assert_eq!(
            json_to_lua_literal(&serde_json::json!(["a", "b"])),
            "{\"a\", \"b\"}"
        );
        assert_eq!(
            json_to_lua_literal(&serde_json::json!({"k": 1})),
            "{[\"k\"]=1}"
        );
    }

    #[test]
    fn extract_prefers_content_then_response_then_whole() {
        // (1) `content` takes priority (DefaultAgent invoker / agent.run return-value path).
        let p = serde_json::json!({
            "content": "Water boils at 100°C",
            "messages": [{"role": "assistant"}],
            "usage": {"input_tokens": 67, "output_tokens": 29},
            "ok": true,
        });
        let (value, ok) = WorkerResultCaptor::extract(&p);
        assert_eq!(value, serde_json::json!("Water boils at 100°C"));
        assert!(ok);

        // (2) No `content` → `response` (caller-script convention worker_result).
        let p = serde_json::json!({ "ok": false, "response": {"patch": "..."} });
        let (value, ok) = WorkerResultCaptor::extract(&p);
        assert_eq!(value, serde_json::json!({"patch": "..."}));
        assert!(!ok);

        // (3) Neither present → the whole payload (custom shape).
        let p = serde_json::json!({ "custom_field": 42 });
        let (value, ok) = WorkerResultCaptor::extract(&p);
        assert_eq!(value, serde_json::json!({"custom_field": 42}));
        assert!(ok); // `ok` absent → defaults to true
    }

    #[tokio::test]
    async fn captor_emits_worker_result_from_payload() {
        let (tx, rx) = oneshot::channel();
        let captor = WorkerResultCaptor {
            tx: Mutex::new(Some(tx)),
            sink: None,
            declared_model: None,
        };
        let payload = serde_json::json!({ "ok": true, "response": "hello" });
        let ack = captor
            .call("worker_result".into(), "evt-1".into(), payload, Value::Null)
            .await
            .expect("handler ack");
        assert_eq!(ack, Value::Null);
        let wr = rx.await.expect("recv");
        assert!(wr.ok);
        assert_eq!(wr.value, serde_json::json!("hello"));
    }

    // ─── declared model → per-step stats sidecar ─────────────────────────

    /// Run one payload through a captor carrying `declared_model` and
    /// return the resulting `stats.model`.
    async fn captured_model(declared: Option<&str>, payload: Value) -> Option<String> {
        let (tx, rx) = oneshot::channel();
        let captor = WorkerResultCaptor {
            tx: Mutex::new(Some(tx)),
            sink: None,
            declared_model: declared.map(str::to_string),
        };
        captor
            .call("agent_result".into(), "evt-1".into(), payload, Value::Null)
            .await
            .expect("handler ack");
        let wr = rx.await.expect("recv");
        wr.stats
            .expect("worker_kind alone guarantees a sidecar")
            .model
    }

    /// `agent.run` reports no model, so the declared `profile.model` is
    /// what lands in the sidecar — the gap that left `StepEntry.stats
    /// .model` permanently `None` on this backend.
    #[tokio::test]
    async fn declared_model_lands_in_the_stats_sidecar() {
        let payload = serde_json::json!({
            "content": "done",
            "usage": {"input_tokens": 10, "output_tokens": 4},
            "num_turns": 2,
        });
        assert_eq!(
            captured_model(Some("opus"), payload).await,
            Some("opus".to_string())
        );
    }

    /// The fallback does not depend on the usage rail: a caller-script
    /// `worker_result` shape (no `usage`, no `num_turns`) still gets the
    /// declared model, because `ensure_worker_kind` has already created
    /// the sidecar.
    #[tokio::test]
    async fn declared_model_lands_even_without_usage_in_the_payload() {
        let payload = serde_json::json!({ "ok": true, "response": "hello" });
        assert_eq!(
            captured_model(Some("sonnet"), payload).await,
            Some("sonnet".to_string())
        );
    }

    /// A runtime-reported `model` is what actually served the attempt, so
    /// it outranks the declaration.
    #[tokio::test]
    async fn runtime_reported_model_wins_over_the_declaration() {
        let payload = serde_json::json!({
            "content": "done",
            "model": "claude-runtime-1",
            "usage": {"input_tokens": 10, "output_tokens": 4},
        });
        assert_eq!(
            captured_model(Some("opus"), payload).await,
            Some("claude-runtime-1".to_string())
        );

        // …including when the payload carries nothing else the stats
        // extractor keys on.
        let payload = serde_json::json!({ "content": "done", "model": "claude-runtime-1" });
        assert_eq!(
            captured_model(Some("opus"), payload).await,
            Some("claude-runtime-1".to_string())
        );
    }

    /// No declaration and no runtime report = no attribution invented.
    #[tokio::test]
    async fn no_declared_model_leaves_the_sidecar_model_empty() {
        let payload = serde_json::json!({ "ok": true, "response": "hello" });
        assert_eq!(captured_model(None, payload).await, None);
    }

    #[tokio::test]
    async fn factory_builds_prompt_based_agent_when_script_path_absent() {
        use crate::blueprint::compiler::SpawnerFactory;
        use crate::blueprint::{AgentDef, AgentKind, AgentProfile};

        let factory = AgentBlockInProcessSpawnerFactory::new();
        let ad = AgentDef {
            name: "writer".into(),
            kind: AgentKind::AgentBlock,
            spec: serde_json::json!({}),
            profile: Some(AgentProfile {
                system_prompt: "You are writer.".into(),
                ..Default::default()
            }),
            meta: None,
            runner: None,
            runner_ref: None,
            verdict: None,
        };
        let _spawner = factory.build(&ad, None).expect("factory build");
        // = ScriptSource::Inline path (self-hosted invoker, mcp_servers embed);
        // the host_handler single sink captures every event kind.
    }

    // ─── GH #86: effective tool grant ─────────────────────────────────────

    fn agent_block_def(name: &str, spec: Value, tools: &[&str]) -> crate::blueprint::AgentDef {
        use crate::blueprint::{AgentDef, AgentKind, AgentProfile};
        AgentDef {
            name: name.into(),
            kind: AgentKind::AgentBlock,
            spec,
            profile: Some(AgentProfile {
                system_prompt: "You are an auditor.".into(),
                tools: tools.iter().map(|t| t.to_string()).collect(),
                ..Default::default()
            }),
            meta: None,
            runner: None,
            runner_ref: None,
            verdict: None,
        }
    }

    #[test]
    fn mcp_tools_of_keeps_only_server_selecting_names() {
        let tools = vec![
            "Read".to_string(),
            "mcp__outline__list_docs".to_string(),
            "WebSearch".to_string(),
        ];
        assert_eq!(mcp_tools_of(&tools), vec!["mcp__outline__list_docs"]);
        assert!(mcp_tools_of(&["Read".to_string()]).is_empty());
    }

    /// PromptBasedAgent mode is where the grant is enforced: only the
    /// `spec.mcp_servers` entries named by the effective set (=
    /// `profile.tools`, which the compiler has already overwritten from a
    /// declared Runner) reach the invoker.
    ///
    /// Enforcement is per **server**, not per tool — granting
    /// `mcp__outline__list_docs` embeds the whole `outline` server, and the
    /// SDK exposes every tool of a connected server to the model.
    #[tokio::test]
    async fn effective_grant_narrows_the_embedded_mcp_servers() {
        use crate::blueprint::compiler::SpawnerFactory;

        let ad = agent_block_def(
            "auditor",
            serde_json::json!({
                "mcp_servers": [
                    {"name": "outline", "command": "outline-mcp", "args": []},
                    {"name": "semantic-scholar", "command": "ss-mcp", "args": []},
                ]
            }),
            &["mcp__outline__list_docs"],
        );

        // The pure resolution the factory performs, asserted directly (the
        // built `Arc<dyn SpawnerAdapter>` is opaque).
        let effective = ad.profile.as_ref().unwrap().tools.clone();
        let servers = resolve_needed_mcp_servers(
            &effective,
            ad.spec["mcp_servers"].as_array().expect("array"),
        );
        let names: Vec<&str> = servers
            .iter()
            .filter_map(|c| c.get("name").and_then(|n| n.as_str()))
            .collect();
        assert_eq!(
            names,
            vec!["outline"],
            "semantic-scholar is declared in spec but not selected by the grant"
        );

        // And the build itself succeeds on this (PromptBased) path.
        AgentBlockInProcessSpawnerFactory::new()
            .build(&ad, None)
            .expect("PromptBasedAgent mode accepts an MCP grant");
    }

    /// ScriptBasedAgent mode cannot enforce an MCP grant (the script drives
    /// its own `mcp.connect`), so declared `mcp__` entries are rejected
    /// rather than silently ignored.
    #[tokio::test]
    async fn script_mode_rejects_a_declared_mcp_grant() {
        use crate::blueprint::compiler::{CompileError, SpawnerFactory};

        let ad = agent_block_def(
            "gate-danger",
            serde_json::json!({ "script_path": "gate.lua" }),
            &["mcp__outline__list_docs"],
        );
        let err = AgentBlockInProcessSpawnerFactory::new()
            .build(&ad, None)
            .err()
            .expect("must reject");
        match err {
            CompileError::InvalidSpec { name, msg } => {
                assert_eq!(name, "gate-danger");
                assert!(msg.contains("mcp.connect"), "explains why: {msg}");
                assert!(
                    msg.contains("PromptBasedAgent"),
                    "names the actionable alternative: {msg}"
                );
            }
            other => panic!("expected InvalidSpec, got: {other:?}"),
        }
    }

    /// The guard is scoped to `mcp__` entries: an empty grant (the issue's
    /// own repro BP) and an inert-only grant (an agent.md `tools: Read,
    /// WebSearch` line, which compiled before the guard existed) both still
    /// build in script mode.
    #[tokio::test]
    async fn script_mode_accepts_empty_and_inert_grants() {
        use crate::blueprint::compiler::SpawnerFactory;

        let spec = serde_json::json!({ "script_path": "gate.lua" });
        for tools in [&[][..], &["Read", "WebSearch"][..]] {
            let ad = agent_block_def("gate-danger", spec.clone(), tools);
            AgentBlockInProcessSpawnerFactory::new()
                .build(&ad, None)
                .unwrap_or_else(|e| panic!("script mode must accept tools {tools:?}: {e}"));
        }
    }

    #[tokio::test]
    async fn factory_builds_script_based_agent_when_script_path_present() {
        use crate::blueprint::compiler::SpawnerFactory;
        use crate::blueprint::{AgentDef, AgentKind, AgentProfile};

        let factory = AgentBlockInProcessSpawnerFactory::new();
        let ad = AgentDef {
            name: "patch-spawner".into(),
            kind: AgentKind::AgentBlock,
            spec: serde_json::json!({
                "script_path": "assets/operator_scripts/blueprint_patch_spawner.lua",
                "project_root": ".",
            }),
            profile: Some(AgentProfile {
                system_prompt: "Patch generator.".into(),
                ..Default::default()
            }),
            meta: None,
            runner: None,
            runner_ref: None,
            verdict: None,
        };
        let _spawner = factory.build(&ad, None).expect("factory build");
        // = ScriptSource::Path path; caller-provided script; host_handler single sink.
    }

    // ─── Issue #17: `project_root` priority chain ─────────────────────────

    #[test]
    fn resolve_spec_project_root_uses_spec_value_when_present() {
        let resolved =
            resolve_spec_project_root(&serde_json::json!({ "project_root": "/spec-root" }));
        assert_eq!(resolved, PathBuf::from("/spec-root"));
    }

    #[test]
    fn resolve_spec_project_root_falls_back_to_env_current_dir_when_spec_absent() {
        let resolved = resolve_spec_project_root(&serde_json::json!({}));
        assert_eq!(
            resolved,
            std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
        );
    }

    /// A view carrying exactly the task-context fields under test. Built
    /// through the real `AgentContextView::from_ctx` so the field names
    /// stay bound to the canonical `ctx.meta.runtime` keys rather than to
    /// a hand-written literal that could drift from them.
    fn view_with(pairs: &[(&str, Value)]) -> AgentContextView {
        let mut ctx = crate::core::ctx::Ctx::new(
            crate::types::StepId::parse("ST-project-root").unwrap(),
            1,
            "writer",
        );
        for (k, v) in pairs {
            ctx.meta.runtime.insert((*k).to_string(), v.clone());
        }
        AgentContextView::from_ctx(&ctx)
    }

    // ─── project_root priority chain (issue #17, now off the seam) ────────

    #[test]
    fn project_root_falls_back_to_spec_when_the_view_carries_neither() {
        let view = view_with(&[]);
        let resolved = resolve_project_root(Some(&view), Path::new("/spec-root"));
        assert_eq!(resolved, PathBuf::from("/spec-root"));
    }

    /// No `Ctx` on the caller path at all (`inv.context == None`) is the
    /// same outcome as an empty view — the compile-time fallback stands.
    #[test]
    fn project_root_falls_back_to_spec_without_a_view() {
        assert_eq!(
            resolve_project_root(None, Path::new("/spec-root")),
            PathBuf::from("/spec-root")
        );
    }

    #[test]
    fn project_root_prefers_the_view_over_spec() {
        let view = view_with(&[(TASK_PROJECT_ROOT_KEY, serde_json::json!("/ctx-root"))]);
        let resolved = resolve_project_root(Some(&view), Path::new("/spec-root"));
        assert_eq!(resolved, PathBuf::from("/ctx-root"));
    }

    #[test]
    fn project_root_prefers_work_dir_over_project_root() {
        let view = view_with(&[
            (TASK_PROJECT_ROOT_KEY, serde_json::json!("/ctx-root")),
            (TASK_WORK_DIR_KEY, serde_json::json!("/ctx-work")),
        ]);
        let resolved = resolve_project_root(Some(&view), Path::new("/spec-root"));
        assert_eq!(resolved, PathBuf::from("/ctx-work"));
    }

    // ─── GH #86: task_metadata delivery via SDK extra_globals ─────────────

    /// An `artifact` emit must reach the sink as an
    /// `OutputEvent::Artifact` AND leave the invocation running, so the
    /// script can stage parts and still finish with its terminal emit.
    #[tokio::test]
    async fn artifact_kind_stages_a_named_part_without_completing() {
        use crate::worker::output::{ContentRef, OutputEvent, OutputSink};

        #[derive(Default)]
        struct RecordingSink(Mutex<Vec<OutputEvent>>);
        #[async_trait]
        impl OutputSink for RecordingSink {
            async fn emit(&self, event: OutputEvent) -> Result<(), crate::EngineError> {
                self.0.lock().unwrap().push(event);
                Ok(())
            }
        }

        let sink = Arc::new(RecordingSink::default());
        let (tx, mut rx) = oneshot::channel();
        let captor = WorkerResultCaptor {
            tx: Mutex::new(Some(tx)),
            sink: Some(sink.clone()),
            declared_model: None,
        };

        captor
            .call(
                ARTIFACT_EVENT_KIND.into(),
                "evt-1".into(),
                serde_json::json!({ "name": "verdict", "content": "PASS" }),
                Value::Null,
            )
            .await
            .expect("staging must succeed");

        let staged = sink.0.lock().unwrap().clone();
        assert_eq!(staged.len(), 1, "exactly one artifact staged");
        match &staged[0] {
            OutputEvent::Artifact { name, content } => {
                assert_eq!(name, "verdict");
                // `ContentRef` is not `PartialEq`; match the variant.
                match content {
                    ContentRef::Inline { value } => {
                        assert_eq!(value, &serde_json::json!("PASS"))
                    }
                    other => panic!("expected Inline content, got: {other:?}"),
                }
            }
            other => panic!("expected Artifact, got: {other:?}"),
        }
        assert!(
            rx.try_recv().is_err(),
            "an artifact emit must NOT complete the invocation"
        );

        // The terminal emit still lands afterwards.
        captor
            .call(
                "worker_result".into(),
                "evt-2".into(),
                serde_json::json!({ "ok": true, "response": "done" }),
                Value::Null,
            )
            .await
            .expect("terminal emit");
        assert_eq!(
            rx.await.expect("recv").value,
            serde_json::json!("done"),
            "the non-reserved kind still completes the invocation"
        );
    }

    /// An `artifact` emit with no `name` cannot address a part, so it is
    /// an error back to the script rather than a silent drop.
    #[tokio::test]
    async fn artifact_without_a_name_is_reported_to_the_script() {
        let (tx, _rx) = oneshot::channel();
        let captor = WorkerResultCaptor {
            tx: Mutex::new(Some(tx)),
            sink: None,
            declared_model: None,
        };
        let err = captor
            .call(
                ARTIFACT_EVENT_KIND.into(),
                "evt-1".into(),
                serde_json::json!({ "content": "PASS" }),
                Value::Null,
            )
            .await
            .expect_err("a nameless artifact must fail loud");
        assert!(
            format!("{err}").contains("name"),
            "names the missing field: {err}"
        );
    }

    // ─── GH #86: the shared view → Lua-global mapping ─────────────────────

    #[test]
    fn context_globals_renders_task_metadata_and_agent_ctx() {
        let mut view = view_with(&[(TASK_METADATA_KEY, serde_json::json!({"issue": 86}))]);
        view.extra.insert(
            "org_conventions".to_string(),
            serde_json::json!("two-space indent"),
        );
        let globals = context_globals(Some(&view));
        assert_eq!(
            globals.get(TASK_METADATA_GLOBAL),
            Some(&serde_json::json!({"issue": 86}))
        );
        assert_eq!(
            globals.get(AGENT_CTX_GLOBAL),
            Some(&serde_json::json!({"org_conventions": "two-space indent"})),
            "Blueprint-declared agent ctx must reach the in-process lane too"
        );
    }

    /// Absent fields contribute no entry, so a script sees `nil` and can
    /// branch on presence — an empty table would be indistinguishable from
    /// "the author declared an empty ctx".
    #[test]
    fn context_globals_omits_absent_fields() {
        assert!(context_globals(None).is_empty(), "no view → no globals");
        assert!(
            context_globals(Some(&view_with(&[]))).is_empty(),
            "empty view → no globals"
        );

        let view = view_with(&[(TASK_METADATA_KEY, serde_json::json!({"issue": 86}))]);
        let globals = context_globals(Some(&view));
        assert!(globals.contains_key(TASK_METADATA_GLOBAL));
        assert!(
            !globals.contains_key(AGENT_CTX_GLOBAL),
            "an empty `extra` must not render an empty _AGENT_CTX table"
        );
    }

    /// Neither global may collide with an SDK-reserved name, and the two
    /// must not collide with each other.
    #[test]
    fn context_globals_use_names_the_sdk_does_not_reserve() {
        for name in [TASK_METADATA_GLOBAL, AGENT_CTX_GLOBAL] {
            for reserved in ["_PROMPT", "_CONTEXT", "_SCRIPT_NAME"] {
                assert_ne!(name, reserved);
            }
        }
        assert_ne!(TASK_METADATA_GLOBAL, AGENT_CTX_GLOBAL);
    }

    /// The global name must not collide with the three the SDK reserves
    /// for itself — a collision would be silently overwritten one way or
    /// the other depending on injection order.
    #[test]
    fn task_metadata_global_does_not_shadow_an_sdk_reserved_name() {
        for reserved in ["_PROMPT", "_CONTEXT", "_SCRIPT_NAME"] {
            assert_ne!(TASK_METADATA_GLOBAL, reserved);
        }
    }

    /// Script mode keeps `ScriptSource::Path`: delivering `task_metadata`
    /// through `extra_globals` means the chunk itself is never rewritten,
    /// so a caller script's own directory stays on `package.path` (sibling
    /// `require` keeps working) and its Lua stack-trace line numbers are
    /// unshifted. The `sibling_require_resolves_*` e2e is the behavioural
    /// half of this claim.
    #[tokio::test]
    async fn script_mode_never_rewrites_the_caller_chunk() {
        use crate::blueprint::compiler::SpawnerFactory;

        let ad = agent_block_def(
            "gate-danger",
            serde_json::json!({ "script_path": "/nonexistent/gate.lua" }),
            &[],
        );
        // A build must succeed without touching the file: the path is
        // handed to the SDK verbatim, exactly as before GH #86.
        AgentBlockInProcessSpawnerFactory::new()
            .build(&ad, None)
            .expect("script mode must not read the script at build time");
    }
}