Skills › Scientific Agent Skills › simpy › bounded_queue_scenario.py
bounded_queue_scenario.py
126 lines · skills/simpy/scripts/bounded_queue_scenario.py at 5f2bc03
1#!/usr/bin/env python3
2"""Run a bounded finite-horizon queue scenario from allowlisted JSON."""
4from __future__ import annotations
6import argparse
7import json
8from pathlib import Path
9from typing import Any
11from _common import (
12 MAX_TRACE_RECORDS,
13 CliError,
14 emit_json,
15 emit_text,
16 integer,
17 load_json_object,
18)
19from basic_simulation_template import QueueConfig, run_simulation
22def _trace_text(records: list[dict[str, Any]]) -> str:
23 return "".join(
24 json.dumps(
25 record,
26 sort_keys=True,
27 ensure_ascii=False,
28 allow_nan=False,
29 separators=(",", ":"),
30 )
31 + "\n"
32 for record in records
33 )
36def run_scenario(
37 config: QueueConfig,
38 *,
39 trace_max_records: int | None = None,
40) -> dict[str, Any]:
41 """Run the built-in loss/queue scenario; no plugin or Python code is loaded."""
43 return run_simulation(
44 config,
45 replication=0,
46 trace_max_records=trace_max_records,
47 )
50def build_parser() -> argparse.ArgumentParser:
51 parser = argparse.ArgumentParser(
52 description=(
53 "Run the built-in bounded M/M/c/K-style queue scenario. The local "
54 "JSON configuration has a fixed allowlist; no code, plugin, eval, "
55 "network resource, or unbounded simulation is accepted."
56 )
57 )
58 parser.add_argument(
59 "--config",
60 help="Optional local .json queue configuration; URLs and symlinks are rejected",
61 )
62 parser.add_argument("--output", help="Optional local .json scenario report")
63 parser.add_argument(
64 "--trace-output",
65 help="Optional local .jsonl deterministic event trace",
66 )
67 parser.add_argument(
68 "--trace-max-records",
69 type=int,
70 default=100_000,
71 help="Trace cap from 1 through 1,000,000 (default: 100000)",
72 )
73 parser.add_argument("--force", action="store_true", help="Replace explicit outputs")
74 return parser
77def main(argv: list[str] | None = None) -> int:
78 parser = build_parser()
79 args = parser.parse_args(argv)
80 try:
81 config = QueueConfig.from_mapping(
82 load_json_object(args.config) if args.config else None
83 )
84 trace_limit = (
85 integer(
86 args.trace_max_records,
87 name="trace_max_records",
88 minimum=1,
89 maximum=MAX_TRACE_RECORDS,
90 )
91 if args.trace_output
92 else None
93 )
94 report = run_scenario(config, trace_max_records=trace_limit)
95 trace = report.pop("event_trace", None)
96 if trace is not None:
97 records = trace["records"]
98 emit_text(
99 _trace_text(records),
100 output=args.trace_output,
101 suffixes={".jsonl"},
102 force=args.force,
103 )
104 report["trace"] = {
105 "file": Path(args.trace_output).name,
106 "records": len(records),
107 "truncated": trace["truncated"],
108 }
109 else:
110 report["trace"] = None
111 report["safety"] = {
112 "config_allowlist": True,
113 "entity_limit": config.max_entities,
114 "event_limit": config.max_events,
115 "network_used": False,
116 "plugin_loading": False,
117 "time_limit": config.horizon,
118 }
119 emit_json(report, output=args.output, force=args.force)
120 except CliError as exc:
121 parser.error(str(exc))
122 return 0
125if __name__ == "__main__":
126 raise SystemExit(main())