File size: 138,190 Bytes
488d96b | 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 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071 2072 2073 2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 2115 2116 2117 2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464 2465 2466 2467 2468 2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493 2494 2495 2496 2497 2498 2499 2500 2501 2502 2503 2504 2505 2506 2507 2508 2509 2510 2511 2512 2513 2514 2515 2516 2517 2518 2519 2520 2521 2522 2523 2524 2525 2526 2527 2528 2529 2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 | """CPython bootstrap for dsh-ptc-runtime-python.
Reads a :class:`BootMessage` on fd 3, applies resource limits and log capture,
reads a :class:`RunMessage`, runs the model program as the body of an async
function (top-level ``await`` and ``return`` both work; the returned value is
the completion), and posts a terminal :class:`DoneMessage`. The program calls
host functions through the ``tools`` (or other namespace) proxy, whose attribute
and subscript access return awaitables that ride binding messages over fd 3.
This module runs under ``python3 -I`` with only ``TMPDIR`` in its environment
and ``sys.path`` containing only its own directory.
"""
from __future__ import annotations
import asyncio
import ast
import io
import json
import math
import os
import re
import resource
import signal
import sys
import threading
import traceback
from decimal import Context, Decimal
# The float encoder must NOT depend on the process-global decimal context: a
# legitimate program may set `getcontext().prec = 2` (silently rounding the
# completion value's digits) or `traps[Inexact] = True` (making the encode
# raise, misclassifying a successful run as an exception). A fixed context with
# prec=28 (more than the 17 significant digits a double needs) makes the
# normalize() spelling decision context-independent.
_FLOAT_CONTEXT = Context(prec=28)
from pathlib import Path
from typing import Any
# ``python3 -I`` (isolated) drops the script directory from ``sys.path`` so
# the sibling ``protocol.py`` is invisible by default. Restore it explicitly
# before importing.
sys.path.insert(0, str(Path(__file__).resolve().parent))
from protocol import PROTOCOL_FD, log_truncation_marker # noqa: E402
# Read size for the async fd-3 reader. One `os.read` returns whatever the pipe
# holds, so this only bounds a single syscall's copy, not a frame: a larger frame
# simply takes more reads. 64 KiB matches the usual pipe capacity.
_READ_CHUNK_BYTES = 65536
# Module-level captures for the done-frame LAST-resort fallback. This bootstrap IS
# ``__main__``, so ``import __main__; __main__.os = ...`` would rebind ``os.write``
# at call time inside ``ProtocolChannel.write_encoded``. These module-level names
# are the RAW primitive for the fallback; they are BOUND INTO ``_run`` LOCALS
# before the program runs (see ``send_done``), which is what makes a one-line
# rebind unable to change which write the fallback uses β the module global here
# is itself reachable by ``__main__._os_write = boom``, so the immunity lives in
# the ``_run`` frame-local binding, not in the module global.
_os_write = os.write
_memoryview = memoryview
# A fixed, pre-encoded done frame for the fallback. It carries no live model
# value, so it can always be written even when a transitive name (a ``_dump_*``
# helper or ``os``) has been rebound and the normal encode/write threw. The
# message is the fixed literal ``<unrenderable>`` β distinct from the failure
# reporter's ``_UNRENDERABLE_DIAGNOSTIC`` text; the host renders the run as an
# exception rather than a worker-exit, which is the honest verdict for a settled
# run whose reporting was sabotaged. The bytes are JSON-valid and
# newline-terminated.
_FALLBACK_DONE_FRAME = b'{"type":"done","error":{"kind":"exception","message":"<unrenderable>"}}\n'
# Code-unit ceiling on the exception class name interpolated into the LAST-resort
# failure diagnostic. A metaclass `__name__` property can return any length, and
# that construction runs outside the guard that would otherwise absorb a
# MemoryError, so the name is sliced before it is copied. Generous enough that no
# real class name is touched.
_MAX_FALLBACK_NAME_CHARS = 200
# Mirror of the host's output-budget/address-space gate (src/index.ts's
# OUTPUT_BUDGET_WORST_CASE_ADDRESS_SPACE_MULTIPLE and INTERPRETER_BASELINE_BYTES),
# re-applied against the EFFECTIVE RLIMIT_AS after inheritance clamping. An astral
# character is one character but ~4 bytes of str storage and ~4 UTF-8 bytes, and
# three such copies are live at the peak β the caller's write argument, the line
# slice or joined pending handed to push, and the encode copy push takes β so a
# budget's worst-case peak is twelve times its byte count; the interpreter's own
# footprint is reserved on top. Kept in sync with the host constants by the shared
# reasoning, not a wire field.
_OUTPUT_BUDGET_WORST_CASE_MULTIPLE = 12
_INTERPRETER_BASELINE_BYTES = 64 * 1024 * 1024
# ---------------------------------------------------------------------------
# Log buffer β Python-side ledger for captured text.
# ---------------------------------------------------------------------------
class LogBuffer:
"""Ordered text capture under one shared byte budget.
Once the budget is exhausted the buffer emits exactly one in-band
truncation marker via ``sink`` and silently drops everything after. The
cap is a blast-radius bound; "how much was lost" intentionally stays
unmeasured.
"""
def __init__(self, max_bytes: int, sink) -> None:
self._max_bytes = max_bytes
# The ledger starts one byte below max_bytes: each entry is charged its
# JSON-string cost plus one separator byte, and the serialized outer
# logs array adds one more byte of envelope (two brackets and n-1 commas
# over n entries' separators), so a result that exactly exhausts the
# ledger serializes to exactly max_bytes; WITHOUT the reserved byte it
# would serialize to max_bytes + 1. Reserving that byte keeps an
# admitted result within the configured cap; the truncation-marker entry
# is envelope, not payload, and rides uncharged (``_max_bytes`` stays the
# configured value for the marker's message text).
self._remaining = max_bytes - 1
self._truncated = False
# True while an `open` (unterminated-flush) entry is being accumulated:
# continuation fragments bill only their CONTENT (no quotes β they ride
# on the first fragment β and no separator), so a merged entry's wire
# cost is billed exactly once, split across its fragments, matching the
# host ledger.
self._open_started = False
# Re-entrant so a caller may hold it across a compound read-modify-write
# (``_LogStream.write`` reads ``remaining`` several times and then calls
# ``push`` while still holding it). One lock is shared by this buffer and
# every stream that funnels into it: model code may start daemon threads
# that keep calling ``print`` after the program body returns, and the
# settlement ``flush_line`` on the main coroutine reads and mutates the
# same ``_pending``/ledger state. Without a shared lock the flush could
# interleave with a concurrent ``write`` β dropping or double-counting a
# line, or costing the ``done`` frame on a mangled ledger. Fixing which
# callable runs (binding ``out_stream.flush_line``) does not fix what it
# reads.
self._lock = threading.RLock()
# ``sink(text, truncated=False)``. The marker is emitted with
# ``truncated=True`` so the host can stop its own capture at the same
# point rather than treating the marker as ordinary program output: the
# two ledgers exhaust independently, and one entry larger than
# ``max_bytes`` sends only the marker while the host budget is still
# nearly empty.
self._sink = sink
@property
def lock(self) -> "threading.RLock":
"""The shared re-entrant lock guarding this ledger and its streams' buffers."""
return self._lock
@property
def remaining(self) -> int:
"""Serialized bytes still admissible; zero once truncated (streams use this to bound their own buffering, where a character count is a valid lower bound)."""
return 0 if self._truncated else self._remaining
def push(self, text: str, open: bool = False) -> None:
with self._lock:
self._push_locked(text, open)
def _push_locked(self, text: str, open: bool = False) -> None:
if self._truncated:
return
# Cheap lower bound FIRST: one char is at least one UTF-8 byte and the
# JSON form adds two quotes plus the separator, so a single print() far
# above the budget truncates without ever encoding it β the full encode
# would allocate a second equally large string and could turn a
# truncatable log into an RLIMIT_AS death.
if (len(text) + 3 if not self._open_started else len(text)) > self._remaining:
self._truncated = True
self._sink(log_truncation_marker(self._max_bytes), truncated=True)
return
# A model print() can emit a lone surrogate; strict UTF-8 throws on it
# here. Replace it rather than escaping it the way :func:`_dump_string`
# preserves one inside a completion VALUE: log text is already a
# truncatable, substituting channel (the byte cap replaces the tail with
# a marker), and the ledger below charges the RAW UTF-8 bytes, which
# would undercharge the six-byte escape by half. Bounded: the text
# passed the length check, so this encodes at most ~4x remaining.
try:
raw = text.encode("utf-8")
except UnicodeEncodeError:
raw = text.encode("utf-8", errors="replace")
text = raw.decode("utf-8")
# Charge the SERIALIZED cost β the JSON string form's bytes plus one
# separator byte β exactly as the host ledger does. Charging the raw
# UTF-8 length instead undercharges control-heavy text, whose JSON
# escaping expands it up to sixfold (a NUL costs one raw byte but six
# as its ``\uXXXX`` escape): a NUL flood sized to fit ``maxLogBytes``
# raw would serialize to roughly six times the shared cap, and the child
# could then die on RLIMIT_AS (reported host-side as ``worker-exit``)
# instead of emitting the truncation marker. The +1 also floors an empty
# entry above zero, so a flood of blank ``print()`` lines exhausts the
# budget instead of emitting unbounded zero-cost log frames.
# Split billing for a merged entry: the FIRST fragment pays the full
# JSON-string cost plus the separator; every later fragment β a
# continuation OR the closing frame (it is the merged entry's tail, not
# a new entry) β pays only its content, since the quotes and separator
# were billed on the first fragment. A standalone closed entry (no open
# in progress) pays the full cost as before.
if self._open_started:
cost = _json_string_cost(raw) - 2
if cost < 0:
cost = 0
else:
cost = _json_string_cost(raw) + 1
if cost > self._remaining:
self._truncated = True
self._sink(log_truncation_marker(self._max_bytes), truncated=True)
return
self._remaining -= cost
if open:
self._open_started = True
else:
self._open_started = False
self._sink(text, open=open)
class _LogStream(io.TextIOBase):
"""A newline-coalescing text stream backed by a :class:`LogBuffer`.
Installed as ``sys.stdout`` / ``sys.stderr`` before executing the model
program. ``print(...)`` calls ``write`` once per argument, separator, and
newline, so a raw one-push-per-write stream would emit
``["a", " ", "b", "\\n"]`` for ``print("a", "b")`` β and PTC mode renders
``logs`` with ``join('\\n')``, turning that into spurious blank lines. This
stream instead buffers writes and pushes one LogBuffer entry per completed
LINE (the text up to each ``\\n``, newline stripped), so the rendered join
reproduces ``a b``. Any unterminated tail is flushed by :meth:`flush_line`
after the program settles.
"""
def __init__(self, logs: LogBuffer) -> None:
super().__init__()
self._logs = logs
# list-of-chunks, joined only at a newline or flush: repeated
# ``print("x", end="")`` must not concatenate quadratically.
self._pending: list[str] = []
self._pending_blocks: list[str] = []
self._pending_chars = 0
# A newline-free drip must not accumulate one list slot per ``write``:
# under a large ``maxLogBytes`` the list-of-fragments pointer array and
# the per-fragment str objects cost host memory well before the byte
# budget is reached, and a 25 M single-character drip would OOM on its
# own accounting (plus the same-size list ``_push_bounded_prefix`` then
# builds). Past this many fragments the chunks are SEALED into a
# ``_pending_blocks`` entry (the character count is unchanged), bounding
# the live fragment count exactly as the host-side ``captureStray`` does
# with its ``MAX_PENDING_CHUNKS``. The seal is INCREMENTAL: only the
# current ``_pending`` fragments (β€ cap) are joined into one block, not
# the whole accumulated buffer, so a large drip stays O(B) rather than
# re-copying the growing block O(BΒ²/cap) times.
self._PENDING_MAX_CHUNKS = 1024
def _pending_parts(self) -> list[str]:
"""The sealed blocks followed by the current fragments, for joining."""
return [*self._pending_blocks, *self._pending]
def writable(self) -> bool: # noqa: D401 -- inherited contract
return True
def write(self, text: str) -> int: # noqa: D401 -- inherited contract
# Serialize the whole read-modify-write against the settlement flush and
# any other thread's write: model code may spawn daemon threads that keep
# printing after the program body returns, and this method reads
# ``remaining`` and mutates ``_pending``/the ledger across many steps. The
# lock is the buffer's and is re-entrant, so the ``push`` calls below
# (which re-acquire it) do not deadlock.
with self._logs.lock:
return self._write_locked(text)
def _write_locked(self, text: str) -> int:
# Drop an empty write instead of buffering it. An empty chunk adds no
# character, so the budget check below can never fire on it:
# ``while True: sys.stdout.write("")`` would append one list slot per
# call with `_pending_chars` pinned at 0, growing unbounded long after
# the log ledger was exhausted (about 3.7 M slots per CPU second here)
# until RLIMIT_AS turned the allocation into a MemoryError β reported as
# the program's own exception rather than the intended bounded-log
# behavior. Returning here also keeps `flush_line` from pushing a
# spurious empty log entry for a program whose only writes were empty.
if not text:
return 0
if "\n" in text:
# Scan `text` in place; the buffered chunks are joined ONLY into the
# first line. Joining the pending chunks with the whole write first
# made a second copy of that write, which an over-budget write
# cannot afford (measured under a 400 MiB addressSpaceMb: one
# buffered character followed by a 340 MiB write died on MemoryError
# inside the join, reported as the program's own exception, and the
# retained chunks made the settlement `flush_line` fail the same way
# β costing the `done` frame and turning the run into a wall-clock
# timeout instead of the promised truncation marker).
length = len(text)
pos = 0
if self._pending or self._pending_blocks:
newline = text.index("\n")
# The +3 cheap-bound overhead (quotes + separator) belongs to a
# NEW entry. While an `open` entry is accumulating, the closing
# line is that entry's TAIL: its cheap bound is the content
# length alone, matching `_push_locked`'s open-aware bound.
# This bound's observable behavior is invariant under the
# overhead either way: when the +3 form trips and the open-aware
# form does not (pending + newline in [remaining - 2, remaining]),
# _push_bounded_prefix re-slices the SAME line text (the extra
# slice `text[:newline]` carries no newline) and _push_locked
# admits it under the same open-aware billing, byte for byte.
# The open-aware form only keeps _push_bounded_prefix's "certain
# to reject" precondition true, which is exactly what the scan
# pre-check below does NOT preserve (its slice carries the
# newline, so push genuinely rejects and truncates).
overhead = 3 if not self._logs._open_started else 0
if self._pending_chars + newline + overhead > self._logs.remaining:
# The reconstructed first line cannot fit the ledger, so
# LogBuffer would reject it whole: copy only the prefix that
# fails its cheap bound and drop the chunks. The slice is
# bounded HERE, not inside the helper: `text[:newline]` on a
# 340 MiB newline-terminated write is the same full copy the
# join was (measured: MemoryError inside `sys.stdout.write`
# under a 400 MiB addressSpaceMb), and `remaining + 4`
# characters are all the helper can use.
self._push_bounded_prefix(text[: min(newline, self._logs.remaining + 4)])
else:
self._pending.append(text[:newline])
line = "".join(self._pending_parts())
self._pending = []
self._pending_blocks = []
self._pending_chars = 0
self._logs.push(line)
pos = newline + 1
# Scan by offset and STOP once the ledger is exhausted: a single
# write of many newlines (``print("\n" * 1000000)``) would otherwise
# re-slice the tail once per line and keep pushing long after
# LogBuffer truncated, burning the CPU budget on discarded lines.
# `remaining` reads 0 the instant the buffer truncates, so the loop
# exits immediately; the unscanned tail is simply dropped.
while pos < length and self._logs.remaining > 0:
newline = text.find("\n", pos)
if newline < 0:
break
# Bound the SLICE the same way LogBuffer bounds the encode: a
# first line far above the ledger would be copied whole before
# push could reject it, and that copy is the allocation an
# over-budget write cannot afford. Copy only a budget-sized
# prefix, which push still rejects on its own cheap bound (the
# prefix is longer than `remaining`), so the marker is emitted
# and the oversized line is never materialized.
overhead = 3 if not self._logs._open_started else 0
if newline - pos + overhead > self._logs.remaining:
self._logs.push(text[pos:pos + self._logs.remaining + 4])
break
self._logs.push(text[pos:newline])
pos = newline + 1
if pos < length:
if self._logs.remaining > 0:
# Buffer only a budget-sized PREFIX of the tail, not the whole
# `text[pos:]`: an early newline followed by a huge unterminated
# tail (`"\n" + "A" * 30 MiB`) would otherwise copy the entire
# tail into `_pending` here β a second full copy of the model's
# own string, the RLIMIT_AS death this path exists to avoid β
# before the newline-free trigger below could bound it. Anything
# past `remaining` characters cannot be admitted (the char count
# is a lower bound on the serialized cost), so a
# `remaining + 4`-character prefix is all that can ever survive;
# the flush trigger below rejects it and emits the marker.
tail = text[pos:pos + self._logs.remaining + 4]
self._pending.append(tail)
self._pending_chars = len(tail)
else:
# The ledger ran out with text still unscanned, so that text
# IS being dropped and the run must say so. One push is
# enough and is bounded: `remaining` is 0, so LogBuffer's
# cheap length lower bound rejects immediately, emits the
# marker, and never encodes the tail β and a push after the
# marker is already out returns without emitting a second.
# Reaching 0 EXACTLY (65 one-character lines against the
# default 3-byte-per-entry serialized charge) leaves
# `_truncated` unset, so without this the tail vanished with
# no marker at all. Sliced to a budget-sized prefix, not the
# whole tail: the tail can be hundreds of megabytes and the
# copy would be the RLIMIT_AS death this bound exists to
# avoid, while push only needs enough characters to fail its
# own cheap length check.
self._logs.push(text[pos:pos + self._logs.remaining + 4])
else:
self._pending.append(text)
self._pending_chars += len(text)
# Seal the fragment list past the chunk cap: a newline-free drip
# appends one fragment per write, so a 25 M single-character flood
# would accumulate that many list slots (and str objects) long before
# the byte budget is met β the pointer array alone being ~25 M slots.
# Past the cap the current fragments are joined into ONE block and
# moved to `_pending_blocks` (character count unchanged), bounding the
# live fragment count exactly as the host-side `captureStray` seal
# does. The join is only the β€cap current fragments, never the whole
# accumulated buffer, so a large drip stays O(B) rather than
# re-copying the growing block O(BΒ²/cap) times; a newline never starts
# a multi-byte sequence, so the fragments are un-sealable mid-line.
if len(self._pending) >= self._PENDING_MAX_CHUNKS:
self._pending_blocks.append("".join(self._pending))
self._pending = []
# A newline-free flood must hit the budget while running, not at
# settlement: once the buffered tail alone can no longer fit the
# ledger (chars lower-bound the serialized cost), push it through β LogBuffer
# truncates, emits the marker once, and swallows everything after.
if self._pending_chars > self._logs.remaining:
self._push_bounded_prefix()
return len(text)
def _push_bounded_prefix(self, extra: str = "") -> None:
# Reached only when the buffered characters already exceed what the
# ledger admits, so LogBuffer is certain to reject on its cheap length
# bound and emit the marker. Copy a budget-sized PREFIX rather than the
# joined whole: ``sys.stdout.write("x")`` followed by one newline-free
# 340 MiB write leaves two chunks whose join is a second copy of the
# payload, and under a tight addressSpaceMb that join raises MemoryError
# from inside `write` β surfacing as the program's own exception, or,
# while the oversized chunks stayed retained, again from `flush_line`
# after the program settled, which cost the `done` frame and turned the
# run into a wall-clock timeout instead of the promised truncation
# marker.
#
# The chunks are dropped BEFORE the push so neither this call nor the
# settlement flush can repeat the allocation, and dropping the text is
# exactly what the marker reports. `remaining + 4` is the shortest
# prefix that still fails LogBuffer's ``len(text) + 3 > remaining``
# check; the accumulation stops there, so the copy is bounded by the log
# budget however large the pending chunks are.
#
# `self._pending` is iterated IN PLACE and `extra` handled after it:
# `(*self._pending, extra)` would first copy every pending reference into
# a same-size tuple, which for a single-character drip (millions of tiny
# chunks) is a second pointer array as large as the list itself --
# measured at +80 MiB of tuple on top of a 40 MiB list for 5.2M chunks,
# the allocation this bounded prefix exists to avoid.
limit = self._logs.remaining + 4
parts: list[str] = []
total = 0
for chunk in self._pending_parts():
parts.append(chunk[: limit - total])
total += len(parts[-1])
if total >= limit:
break
else:
# Only reached when the pending chunks did not fill the prefix, so
# `extra` is the one remaining source of text.
parts.append(extra[: limit - total])
self._pending = []
self._pending_blocks = []
self._pending_chars = 0
self._logs.push("".join(parts))
def flush(self) -> None: # noqa: D401 -- inherited contract
# ``TextIOBase.flush`` is a no-op, so without this override an explicit
# ``print(..., flush=True)`` or ``sys.stdout.flush()`` left the text in
# `_pending` with nothing to drain it except `flush_line` after the
# program settles. A run that then hangs or is killed never reaches that
# call: ``print("before hang", end="", flush=True)`` followed by an
# infinite loop returned `logs: []`, losing the one diagnostic the
# program deliberately committed. Forwarding makes an explicit flush emit
# the pending entry immediately, which is what the caller asked for; a
# newline-terminated write already emitted on its own.
self.flush_line()
def flush_line(self) -> None:
"""Push any buffered text not terminated by a newline (also serves explicit flushes)."""
# Same shared, re-entrant lock as ``write``: the settlement flush on the
# main coroutine and a daemon thread's concurrent ``write`` both touch
# ``_pending`` and the ledger, so this read-and-clear must be atomic
# against them.
with self._logs.lock:
if self._pending or self._pending_blocks:
# Join, drop the chunks, THEN push β the same join-clear-push order
# as `_write_locked`'s newline branch. Pushing before the clear would keep the
# pending chunks alive through `_push_locked`'s `text.encode`, so
# the chunks, their join, and the encode copy would all be live at
# once; dropping the chunks first leaves only the join and its
# encode, matching that path's peak.
line = "".join(self._pending_parts())
self._pending = []
self._pending_blocks = []
self._pending_chars = 0
# The line has NO trailing newline: mark the frame `open` so the
# host appends the next log frame to the same entry instead of
# inserting a fake newline between two entries.
self._logs.push(line, open=True)
# ---------------------------------------------------------------------------
# Fd-3 channel β line-framed JSON.
# ---------------------------------------------------------------------------
# Non-string scalars only. The string form is scanned by hand in
# :func:`_decode_json_plain` because a ``(?:[^"\\]|\\.)*`` repetition makes
# CPython's backtracking engine retain per-repetition state proportional to the
# string's WIDTH: measured at ~146 MiB of engine state for a 1 MiB string and
# ~558 MiB for 4 MiB, so a legitimate multi-megabyte binding reply raised
# MemoryError out of ``_pump_replies``, leaving its future unsettled until the
# wall clock reported a timeout.
_SCALAR_RE = re.compile(
r'-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?|true|false|null'
)
# A run of ordinary string body characters. The star applies to a CHARACTER
# CLASS, which the engine matches in one linear pass with no backtracking state,
# so the scanner's cost is the number of escapes, not the string's width.
_STRING_CHUNK_RE = re.compile(r'[^"\\]*')
def _decode_json_plain(
text: str,
# Def-time captures: the decoder runs AFTER the program (which is `__main__`)
# may have rebound `__main__.json`, `__main__._SCALAR_RE`,
# `__main__._STRING_CHUNK_RE`, or `__main__.len`; a call-time lookup would
# let a one-line rebind kill the reply pump (a broken decode strands every
# pending Future to the wall clock). Defaults are evaluated at def time.
_json_loads: Any = json.loads,
_scalar_re: Any = _SCALAR_RE,
_string_chunk_re: Any = _STRING_CHUNK_RE,
_len: Any = len,
_isinstance: Any = isinstance,
_str: Any = str,
_list: Any = list,
) -> Any:
"""Parse one JSON document iteratively (no per-level recursion).
``json.loads`` recurses per nesting level and raises ``RecursionError``
around ~10k levels, but a binding reply is depth-unbounded by the seam
contract β the host's iterative encoder happily produces documents
``json.loads`` cannot read back. Scalars (numbers, strings with escapes)
are delegated to ``json.loads`` one token at a time, so their grammar and
semantics stay CPython's own; only the container structure is parsed here
with an explicit stack. Raises ``ValueError`` on malformed input; frames
come from the TRUSTED host, so strictness mirrors ``json.loads`` without
extra hostile-input hardening.
"""
length = _len(text)
def skip_ws(i: int) -> int:
while i < length and text[i] in " \t\n\r":
i += 1
return i
def scan_string(i: int) -> int:
# Walk chunk by chunk: each match consumes every character up to the next
# quote or backslash, so an escape costs one extra step and a plain body
# costs one pass. Returns the offset just past the closing quote.
j = i + 1
while True:
j = _string_chunk_re.match(text, j).end()
if j >= length:
raise ValueError(f"unterminated string at offset {i}")
char = text[j]
if char == '"':
return j + 1
# text[j] is a backslash: skip it and the character it escapes. A
# trailing backslash runs j past `length`, caught on the next pass.
j += 2
def scalar(i: int):
if i < length and text[i] == '"':
end = scan_string(i)
return _json_loads(text[i:end]), end
match = _scalar_re.match(text, i)
if match is None:
raise ValueError(f"invalid JSON at offset {i}")
return _json_loads(match.group(0)), match.end()
def string_key(i: int):
key, end = scalar(i)
if not _isinstance(key, _str):
raise ValueError(f"object key must be a string at offset {i}")
end = skip_ws(end)
if end >= length or text[end] != ":":
raise ValueError(f"expected ':' at offset {end}")
return key, skip_ws(end + 1)
# Frames: a list, or (dict, pending key). `value`/`have_value` carry each
# completed value up to its parent frame.
stack: list[Any] = []
value: Any = None
have_value = False
i = skip_ws(0)
while True:
if not have_value:
ch = text[i] if i < length else ""
if ch == "[":
i = skip_ws(i + 1)
if i < length and text[i] == "]":
i += 1
value, have_value = [], True
else:
stack.append([])
continue
elif ch == "{":
i = skip_ws(i + 1)
if i < length and text[i] == "}":
i += 1
value, have_value = {}, True
else:
key, i = string_key(i)
stack.append(({}, key))
continue
else:
value, i = scalar(i)
have_value = True
if not stack:
i = skip_ws(i)
if i != length:
raise ValueError(f"trailing data at offset {i}")
return value
top = stack[-1]
i = skip_ws(i)
ch = text[i] if i < length else ""
if _isinstance(top, _list):
top.append(value)
if ch == ",":
i = skip_ws(i + 1)
have_value = False
elif ch == "]":
i += 1
stack.pop()
value = top
else:
raise ValueError(f"expected ',' or ']' at offset {i}")
else:
container, key = top
container[key] = value
if ch == ",":
key, i = string_key(skip_ws(i + 1))
stack[-1] = (container, key)
have_value = False
elif ch == "}":
i += 1
stack.pop()
value = container
else:
raise ValueError(f"expected ',' or '}}' at offset {i}")
class ProtocolChannel:
"""Blocking readers and synchronous writers over the fd-3 protocol pipe.
Writes are unbuffered and go straight to the fd, so ``send_sync`` is safe
from inside model code (which may run outside an asyncio task) and from
background tasks alike. Concurrent writers are serialized by ``_write_lock``
around a full-write loop (see ``send_sync``): ``os.write`` releases the GIL,
a frame may exceed ``PIPE_BUF`` (logs up to ``maxLogBytes``, completions up
to ``maxValueBytes``, uncapped ``call`` args), and one ``os.write`` may
consume only part of a frame β so neither the GIL nor per-frame atomicity is
relied on for framing.
"""
def __init__(self, fd: int) -> None:
self._fd = fd
# Residual bytes read past a frame's newline, shared by the blocking and
# async readers. Held here, not in the reading coroutine: the reply pump
# is cancelled once `done` is posted, and read-ahead sitting in a local
# would be lost with it. Both readers use `os.read(self._fd, ...)`
# directly, so no buffered file object wraps the fd.
self._pending = bytearray()
# Serializes writers: os.write releases the GIL, and a frame larger
# than PIPE_BUF is neither atomic nor guaranteed fully consumed by one
# call β without the lock, model-created threads printing while a big
# completion frame drains could interleave bytes mid-frame.
self._write_lock = threading.Lock()
def read_frame(
self,
# Def-time captures for the decode primitives (see _decode_json_plain):
# this runs before the program, but the reply path does not β a rebind
# of `__main__._decode_json_plain`/`__main__.os`/`__main__._READ_CHUNK_BYTES`
# must not break the pump.
_decode: Any = _decode_json_plain,
_os_read: Any = os.read,
_read_chunk: int = _READ_CHUNK_BYTES,
_bytes: Any = bytes,
_len: Any = len,
) -> dict[str, Any] | None:
"""Read one JSON-line frame (iteratively decoded). ``None`` on EOF.
Blocking. Used for the two frames read BEFORE the model program starts
(``boot`` and ``run``), where blocking is what the handshake wants. Reply
frames arriving during the program go through :meth:`read_frame_async`,
which must not occupy a thread.
Reads in CHUNKS into the shared ``_pending`` buffer rather than through
``FileIO.readline()``: the fd is unbuffered (``buffering=0``), so
``readline`` issues one ``os.read(1)`` per byte, and a multi-megabyte
``run`` frame β RLIMIT_CPU already in force by then β would burn the
budget in millions of syscalls before ``ast.parse`` even runs. The chunk
reads and the same residual buffer the async path uses keep read-ahead
past a newline for the next frame.
"""
# Scan only the bytes not yet examined: `find` from a running offset so a
# frame arriving in N chunks costs one linear pass total, not one rescan
# of the whole buffer per chunk (which is quadratic in the frame size).
scanned = 0
while True:
newline = self._pending.find(b"\n", scanned)
if newline >= 0:
line = _bytes(self._pending[:newline])
del self._pending[: newline + 1]
return _decode(line.decode("utf-8"))
scanned = _len(self._pending)
chunk = _os_read(self._fd, _read_chunk)
if not chunk:
# EOF before a newline: drop the partial line, as the host drops
# a frame that never completed.
return None
self._pending.extend(chunk)
async def read_frame_async(
self,
# Def-time captures, same rationale as read_frame.
_decode: Any = _decode_json_plain,
_os_read: Any = os.read,
_read_chunk: int = _READ_CHUNK_BYTES,
_bytes: Any = bytes,
_get_event_loop: Any = asyncio.get_event_loop,
_len: Any = len,
) -> dict[str, Any] | None:
"""Await one JSON-line frame without occupying a thread. ``None`` on EOF.
``loop.run_in_executor(None, read_frame)`` was the obvious spelling and
the wrong one: the default executor spins up its first thread the moment
the program awaits a binding, and on Linux/glibc that thread's 8 MiB
stack plus a 64 MiB per-thread malloc arena reservation are charged to
``RLIMIT_AS`` β measured, the child's mappings went from 30.34 MiB to
102.39 MiB across one ``await tools.*``. Since the limit is already in
force, that ~72 MiB comes straight out of the run's ``addressSpaceMb``:
under a small limit the thread cannot start at all and a legitimate
binding call hangs to ``maxWallMs``, and under a larger one an allocation
that should have fit dies as ``MemoryError``. This is the same accounting
the settlement-time CPU recheck was designed around, where a sampling
thread cost the same 72 MiB.
`loop.add_reader` watches the fd instead, so no thread exists.
Bytes past a frame's newline belong to the next frame, so the residual
lives on the CHANNEL rather than in this coroutine: the pump is cancelled
once ``done`` is posted, and a local buffer would discard whatever it had
read ahead.
"""
loop = _get_event_loop()
# Scan only the not-yet-examined bytes (running offset), so a frame
# arriving across many reads costs one linear pass, not a quadratic
# rescan of the whole buffer per read.
scanned = 0
while True:
newline = self._pending.find(b"\n", scanned)
if newline >= 0:
line = _bytes(self._pending[:newline])
del self._pending[: newline + 1]
return _decode(line.decode("utf-8"))
scanned = _len(self._pending)
ready = loop.create_future()
# `add_reader` only reports readability; the read itself happens here,
# and `os.read` returns whatever is buffered without waiting for more.
loop.add_reader(self._fd, lambda: ready.done() or ready.set_result(None))
try:
await ready
finally:
loop.remove_reader(self._fd)
chunk = _os_read(self._fd, _read_chunk)
if not chunk:
# EOF. Any partial line is dropped, matching how the host drops a
# frame that never completed.
return None
self._pending.extend(chunk)
def send_sync(self, message: dict[str, Any]) -> None:
"""Post one frame synchronously.
Encoded with the iterative :func:`_encode_json_plain` (not
``json.dumps``, whose per-level recursion would raise
``RecursionError`` on a deeply nested completion or call argument the
depth-unbounded ``PtcJsonValue`` contract admits). NaN/Infinity still
raise ``ValueError`` β they would serialize as non-standard tokens
that Node's ``JSON.parse`` rejects, silently dropping the frame, and a
call would then hang until the wall clock instead of failing fast.
Callers turn the ``ValueError`` into their own contract error
(dispatch raises the lossless-JSON message).
"""
self.write_encoded(_encode_json_plain(message))
def write_encoded(self, frame: str) -> None:
"""Write a frame that is ALREADY encoded to its JSON string form.
Appends the frame's trailing newline and full-write-loops the bytes
under the writer lock, identically to :meth:`send_sync`. The consumer
supplies the encoded JSON (a ``"done"`` frame carrying a completion
value that was serialized at its validation point β see
:func:`_done_with_value`); the channel does not re-encode it, so the
bytes written are exactly what was validated with no second traversal
of a live object.
"""
payload = (frame + "\n").encode("utf-8")
# Full-write loop under the writer lock (same rationale as send_sync):
# one os.write may consume only part of a frame beyond PIPE_BUF, and a
# partial or interleaved frame is dropped host-side as malformed JSON.
with self._write_lock:
view = memoryview(payload)
while view:
view = view[os.write(self._fd, view):]
# ---------------------------------------------------------------------------
# Tools proxy β turns ``await tools.name(args)`` into a fd-3 call frame.
# ---------------------------------------------------------------------------
class _Namespace:
"""A proxy for one binding namespace: every declared name routes to the bridge.
Names arrive from :class:`BootMessage.namespaces`. Both attribute access
(``tools.name``) and subscript access (``tools["my-tool"]`` β the SDK's
escape hatch for exotic or reserved names, which are legal function names
on the wire) return a coroutine factory that posts a ``call`` frame and
awaits the matching ``reply``. An undeclared name raises ``AttributeError``
(attribute) or ``KeyError`` (subscript), matching the worker backend's
own-property discipline.
``__getattribute__`` (not ``__getattr__``) intercepts attribute access so a
declared name ALWAYS reaches the bridge β even one that collides with an
inherited attribute like ``__class__``, which ordinary lookup would resolve
on ``object`` before ``__getattr__`` ever ran. Internal state lives under
name-mangled ``_Namespace__*`` attributes; a declared binding with such a
name still wins (declared-names check runs first).
"""
def __init__(self, global_name: str, names: list[str], dispatch) -> None:
self.__global = global_name
self.__names = set(names)
self.__dispatch = dispatch
def __call_for(self, name: str):
dispatch = object.__getattribute__(self, "_Namespace__dispatch")
global_name = object.__getattribute__(self, "_Namespace__global")
async def call(args: Any) -> Any:
return await dispatch(global_name, name, args)
return call
def __getattribute__(self, name: str):
# Declared names route to the bridge unconditionally β before Python
# can resolve an inherited attribute (``__class__``) or our own
# internals. Everything else falls through to normal lookup so the
# proxy machinery itself keeps working.
names = object.__getattribute__(self, "_Namespace__names")
if name in names:
return object.__getattribute__(self, "_Namespace__call_for")(name)
return object.__getattribute__(self, name)
def __getattr__(self, name: str):
# Reached only when normal lookup found nothing (declared names were
# already intercepted above), so this is always an undeclared tool.
raise AttributeError(
f"tool {name!r} is not declared in namespace "
f"{object.__getattribute__(self, '_Namespace__global')!r}"
)
def __getitem__(self, name: str):
names = object.__getattribute__(self, "_Namespace__names")
if name not in names:
raise KeyError(
f"tool {name!r} is not declared in namespace "
f"{object.__getattribute__(self, '_Namespace__global')!r}"
)
return object.__getattribute__(self, "_Namespace__call_for")(name)
class _BindingRejection(Exception):
"""Internal reply-pump rejection, converted by ``dispatch`` into the
namespace's declared error class (or ``RuntimeError``) so the marker type
itself never reaches model code."""
def _make_error_class(
name: str,
member_name_property: str,
# Def-time captures (see the __init__ body): the minted class runs AFTER
# the program, so `__main__.Exception`/`__main__.setattr` rebinds must not
# break the rejection constructor.
_Exception_init: Any = Exception,
_setattr: Any = setattr,
) -> type:
"""Mint one program-visible rejection class per the seam's
``PtcBindingErrorClass`` contract: instances carry the failed member name
under ``member_name_property`` and render as their message."""
def __init__(self, member_name: str, message: str) -> None: # noqa: N807
# Exception.__init__ and setattr are captured as defaults at def time:
# the class constructor runs model-visible code paths, and a rebind of
# `__main__.Exception`/`__main__.setattr` must not break a rejection.
_Exception_init.__init__(self, message)
_setattr(self, member_name_property, member_name)
return type(name, (Exception,), {"__init__": __init__})
def _clamped(which: int, soft: int, hard: int) -> tuple[int, int]:
"""Bound a requested (soft, hard) rlimit pair by BOTH inherited limits.
An unprivileged process may lower a hard limit but never raise it, so a
harness already started under a tighter ceiling (``ulimit -v`` below
``addressSpaceBytes``, or a CPU cap below ``cpuSeconds`` + 1) would make
``setrlimit`` raise ``ValueError`` and fail every run β despite the
inherited limit being STRONGER than the one requested. Clamping keeps the
stricter of the two, which still satisfies the containment contract.
Both inherited bounds matter, not just the hard one. A deployment that
inherited a soft limit BELOW what is requested (e.g. inherited ``(100, 200)``,
requested ``(150, 160)``) must keep the stricter soft β returning the
requested ``150`` would RAISE the effective soft limit, loosening RLIMIT_AS
memory or deferring the RLIMIT_CPU SIGXCPU, the opposite of "strictest of
configured and inherited". So each side is clamped against its inherited
counterpart. ``RLIM_INFINITY`` compares as -1, so an infinite inherited bound
imposes no ceiling and the requested value stands.
"""
inherited_soft, inherited_hard = resource.getrlimit(which)
clamped_soft = soft if inherited_soft == resource.RLIM_INFINITY else min(soft, inherited_soft)
clamped_hard = hard if inherited_hard == resource.RLIM_INFINITY else min(hard, inherited_hard)
# setrlimit requires soft <= hard. Clamping the two sides independently can
# invert them (a finite inherited soft below the clamped hard is fine, but a
# requested hard below the inherited soft would leave soft > hard), so pin
# soft under hard as the final step; the stricter hard ceiling wins.
result_soft = min(clamped_soft, clamped_hard)
result_hard = clamped_hard
# For RLIMIT_CPU, a soft limit EQUAL to the hard limit leaves the kernel no
# window to send SIGXCPU: it checks the hard limit in the same tick and
# SIGKILLs directly (a `ulimit -t N` sets both, and a busy loop then dies by
# SIGKILL, not SIGXCPU). The host classifies a CPU overrun ONLY on ``signal
# === 'SIGXCPU'``, so a definite budget exhaustion would be misreported as a
# `worker-exit`. Lowering the soft limit one unit below the hard (when the
# hard is at least 2, so soft stays positive) keeps the stricter-of-the-two
# containment semantics while giving SIGXCPU a window to fire β the CPU
# overrun is then reported as a timeout, not a worker-exit. This is scoped to
# RLIMIT_CPU: for RLIMIT_AS a one-byte soft differential would only misalign
# the child's applied limit with the host-side budget gate, with no signal to
# preserve. The ``hard >= 2`` guard leaves the ``hard == 1`` blind spot
# (a 1-second dual limit cannot lower soft to 0) β a definite CPU overrun
# there is still reported as `worker-exit`; see the settlement note.
if (
which == resource.RLIMIT_CPU
and result_soft == result_hard
and result_hard >= 2
):
result_soft = result_hard - 1
return (result_soft, result_hard)
# ---------------------------------------------------------------------------
# Main.
# ---------------------------------------------------------------------------
async def _run(channel: ProtocolChannel) -> None:
# The exception class every `except` clause in this function catches is bound
# into a LOCAL at the very top, before any model code runs. This bootstrap IS
# `__main__`, so `__main__.BaseException = RuntimeError` would otherwise
# rebind the module global the `except BaseException` clauses resolve at
# runtime β and a program exception that no longer matches would escape `_run`
# with no done frame, misreporting the run as a `worker-exit`. A frame local
# is not reachable by `__main__._X = ...`, so the catch is immune.
_BaseException = BaseException
# The child inherits the host's SIGXCPU disposition and signal mask. If
# the host ignores or blocks SIGXCPU, the soft RLIMIT_CPU fires but cannot
# stop the child β the hard limit's SIGKILL then classifies a definite CPU
# overrun as substrate death (worker-exit) instead of a timeout. Reset to
# the default disposition and unblock HERE, before the resource-limit setup
# and the boot-namespace construction (which can burn CPU): a huge
# namespace under an inherited ignore would otherwise reach the hard limit
# inside that window. The settle-time enforcer already restores SIG_DFL for
# a program that traps or masks the signal mid-run; this closes the
# inherited-state gap.
signal.signal(signal.SIGXCPU, signal.SIG_DFL)
if getattr(signal, "pthread_sigmask", None) is not None:
signal.pthread_sigmask(signal.SIG_UNBLOCK, (signal.SIGXCPU,))
# `RuntimeError` and the `_BindingRejection` marker class are likewise bound
# into locals: `dispatch`'s `call_failure` and its `except` clause resolve
# them at call time, and the program (running as `__main__`) can rebind the
# module globals β `__main__._BindingRejection = ValueError` would leak the
# marker type into model code, violating the class's conversion contract.
# The names differ from the module globals (`_RuntimeError_cls`) so the
# assignment RHS resolves the module global, not an unbound local.
_RuntimeError_cls = RuntimeError
_BindingRejection_cls = _BindingRejection
# `str` for dispatch's rejection conversion is likewise bound: a program
# rebinding `__main__.str` would otherwise run a hostile callable when the
# binding-rejection message is formatted. `isinstance` for `send_done`'s
# frame-shape check is bound the same way.
_str = str
_isinstance = isinstance
# dispatch's argument-validation, event-loop, and frame-send primitives are
# bound here, before the program runs: a rebind of `__main__._lossless_json_violation`,
# `__main__.asyncio`, or the channel's send method must not turn a legitimate
# binding call into an exception or a wall-clock timeout.
_lossless_json_violation_cls = _lossless_json_violation
_get_event_loop_cls = asyncio.get_event_loop
_send_sync_cls = channel.send_sync
# The frame WRITE primitives are bound too: send_sync's body resolves
# `_encode_json_plain` (module global) and `self.write_encoded` (class
# attribute) at call time, so a program rebinding either before the first
# binding call could turn a legitimate call into an exception. dispatch
# and the log sink both write through these directly (the sink does NOT go
# through send_sync, whose body resolves the same names at call time).
_write_encoded_cls = channel.write_encoded
_encode_plain_cls = _encode_json_plain
# 1. Boot handshake.
boot = channel.read_frame()
if boot is None or boot.get("type") != "boot":
raise RuntimeError("bootstrap: expected boot frame on fd 3")
# A limit that cannot be applied must fail the run as a diagnosable done
# frame, not a bare traceback + exit(1): running the program UNCAPPED would
# silently void the containment contract, and the host can only relay what
# rides the protocol.
try:
# SIGXCPU's default disposition (how the soft CPU limit stops the child)
# dumps core, and the child inherits the host's RLIMIT_CORE β a CPU
# timeout would otherwise write a large memory-bearing core file into
# the workspace. Forbid core dumps first so the timeout path leaves none.
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
# Soft limit at cpuSeconds fires SIGXCPU (its default disposition
# terminates the child; the host classifies that close as a timeout).
# Hard limit at +1s is a SIGKILL backstop for a program that traps
# SIGXCPU and keeps burning CPU.
cpu_soft, cpu_hard = _clamped(
resource.RLIMIT_CPU, boot["cpuSeconds"], boot["cpuSeconds"] + 1
)
resource.setrlimit(resource.RLIMIT_CPU, (cpu_soft, cpu_hard))
# Darwin maps the multi-GB dyld shared cache into every process at
# exec, so any practical RLIMIT_AS cap sits below current usage and
# the kernel rejects it β the child would die here on every run. Skip
# the address-space cap there; RLIMIT_CPU and the host's wall-clock
# ceiling still bound the run.
if sys.platform != "darwin":
addr_bytes = int(boot["addressSpaceBytes"])
effective_as = _clamped(resource.RLIMIT_AS, addr_bytes, addr_bytes)
resource.setrlimit(resource.RLIMIT_AS, effective_as)
# The host rejected an output budget too large for the CONFIGURED
# addressSpaceMb, but a launch environment can inherit a STRICTER
# RLIMIT_AS (e.g. a `ulimit -v` wrapper below addressSpaceMb), which
# `_clamped` correctly lowers the effective limit to β leaving the
# budgets validated against a ceiling the child never gets. Re-check
# both budgets against the EFFECTIVE soft limit here, mirroring the
# host gate (each budget times the worst-case Unicode multiple must
# fit the room left after the interpreter baseline), and fail loud at
# boot rather than letting a near-budget output OOM mid-run. The
# constants match src/index.ts's OUTPUT_BUDGET_WORST_CASE_ADDRESS_
# SPACE_MULTIPLE and INTERPRETER_BASELINE_BYTES.
# `effective_soft` is always finite, so the re-check is
# unconditional: `_clamped` was asked for the finite `addr_bytes` on
# both sides, and each of its branches returns either that value or a
# `min` with an inherited bound -- RLIM_INFINITY is not reachable. A
# guard here would have silently skipped the whole re-check on the
# branch it claimed to protect.
effective_soft = effective_as[0]
budgetable = effective_soft - _INTERPRETER_BASELINE_BYTES
for _budget_key in ("maxLogBytes", "maxValueBytes"):
if int(boot[_budget_key]) * _OUTPUT_BUDGET_WORST_CASE_MULTIPLE >= budgetable:
raise ValueError(
"config.%s is too large for the inherited RLIMIT_AS of %d bytes "
"(a near-budget output would breach it during encode); "
"lower the budget or raise the inherited address-space limit"
% (_budget_key, effective_soft)
)
except _BaseException as exc: # noqa: BLE001 -- report every failure to host; `_BaseException` is a pre-program local
channel.send_sync(
{
"type": "done",
"error": {
"kind": "exception",
# Exception-only rendering: format_exc() would embed the
# absolute installed bootstrap.py path in model-visible
# durable output, leaking host paths into transcripts.
"message": "bootstrap: applying resource limits failed\n"
+ "".join(
traceback.format_exception_only(type(exc), exc)
),
},
}
)
return
logs = LogBuffer(
int(boot["maxLogBytes"]),
# The sink writes through the def-time bound encode+write primitives
# (not _send_sync_cls, whose body still resolves _encode_json_plain and
# self.write_encoded at call time) so a rebind cannot break a log frame.
sink=lambda text, truncated=False, open=False: _write_encoded_cls(
_encode_plain_cls(
{
"type": "log",
"text": text,
**({"truncated": True} if truncated else {}),
**({"open": True} if open else {}),
}
)
),
)
# 2. Wire the tools proxies and the ack.
#
# Each entry records the reply Future AND the loop it was created on. Model
# code may call a binding from a THREAD it started, spelled
# ``asyncio.run(tools.x(...))`` or its own new loop in that thread, so a
# Future here can belong to a loop other than the one ``_pump_replies`` runs
# on. ``asyncio.Future`` is not thread-safe: completing it from another
# thread does not wake its own loop, so the pump schedules the completion on
# the owning loop via ``call_soon_threadsafe`` (see ``_pump_replies``) rather
# than calling ``set_result`` directly.
pending: dict[int, tuple[asyncio.AbstractEventLoop, asyncio.Future[Any]]] = {}
next_id = 0
# Serializes the id claim + write + counter advance in ``dispatch`` against
# both other binding-calling threads and the pump's ``pop``. ``dispatch`` may
# run concurrently on several loops/threads, and the host answers a ``call``
# only when its id is the exact successor of the last one β so ids must reach
# the wire in the order they are claimed. Holding this lock across the write
# (not just the counter arithmetic) is what keeps two threads' frames from
# interleaving on fd 3 out of id order, which the host would reject.
pending_lock = threading.Lock()
error_classes: dict[str, type] = {}
async def dispatch(global_name: str, name: str, args: Any) -> Any:
nonlocal next_id
error_class = error_classes.get(global_name)
def call_failure(message: str) -> BaseException:
# The namespace's declared rejection contract (e.g. PTC mode's
# ToolCallError with .toolName) when present; RuntimeError keeps
# the pre-errorClass behavior for namespaces that declared none.
if error_class is not None:
return error_class(name, message)
return _RuntimeError_cls(message)
# Validate the argument shape before claiming an id, so a rejected call
# leaves no gap in the sequence the host checks. json.dumps would coerce
# a non-string dict key or non-finite float rather than raise (allow_nan
# is off, but key coercion still slips through), silently corrupting what
# the tool receives. Reject up front through the call's error contract.
violation = _lossless_json_violation_cls(args)
if violation is not None:
raise call_failure(f"binding arguments must be lossless JSON ({violation})")
# Ids are consecutive from 0 with NO gaps: the host answers a `call` only
# when its id is the exact successor of the last one, which bounds the
# state it retains to a single number. A frame that never reaches the
# host must therefore not consume an id, so the counter advances only
# once the write has succeeded.
#
# The whole claim-write-advance runs under ``pending_lock`` because a
# binding may be called from more than one thread/loop at once (the model
# can start a thread that runs ``asyncio.run(tools.x(...))``). Without
# the lock two callers could claim the same id, or write their frames to
# fd 3 in an order that does not match their ids β either of which the
# host rejects as an out-of-sequence call. The Future's own loop is
# captured here so ``_pump_replies`` can complete it thread-safely.
loop = _get_event_loop_cls()
with pending_lock:
call_id = next_id
fut: asyncio.Future[Any] = loop.create_future()
pending[call_id] = (loop, fut)
try:
_write_encoded_cls(
_encode_plain_cls(
{
"type": "call",
"id": call_id,
"global": global_name,
"name": name,
"args": args,
}
)
)
except (TypeError, ValueError) as exc:
pending.pop(call_id, None)
raise call_failure(
f"binding arguments must be lossless JSON: {exc}"
) from exc
next_id += 1
try:
return await fut
except _BindingRejection_cls as exc:
raise call_failure(_str(exc)) from None
namespaces: dict[str, Any] = {}
for entry in boot["namespaces"]:
namespaces[entry["global"]] = _Namespace(
entry["global"], entry["names"], dispatch
)
declared = entry.get("errorClass")
if declared:
error_class = _make_error_class(
declared["name"], declared["memberNameProperty"]
)
error_classes[entry["global"]] = error_class
# The class is program-visible under its own name so model code
# can `except ToolCallError as e:` and read the member property.
namespaces[declared["name"]] = error_class
channel.send_sync({"type": "boot-ack"})
# 3. Start a reply-pump task before the run message: replies can arrive
# interleaved with the run's own binding traffic.
# The pump's frame reader is bound here, before the program runs: the
# pump itself starts AFTER the program's top-level statements (no suspension
# point between create_task and `await __dsh_main__`), so a body-local
# `channel.read_frame_async` lookup would resolve a rebound class method.
pump_read = channel.read_frame_async
reply_task = asyncio.get_event_loop().create_task(
_pump_replies(channel, pending, pending_lock, pump_read)
)
# 4. Read the run message.
run = channel.read_frame()
if run is None or run.get("type") != "run":
reply_task.cancel()
raise RuntimeError("bootstrap: expected run frame on fd 3")
program: str = run["program"]
# 5. Install log capture β ``print``, tracebacks, and ordinary ``sys.stdout``
# writes funnel into the LogBuffer. The real fds stay open (host uses
# stderr for stray-byte accounting) but the Python-visible streams point
# at the buffer.
sys.stdout = _LogStream(logs) # type: ignore[assignment]
sys.stderr = _LogStream(logs) # type: ignore[assignment]
out_stream, err_stream = sys.stdout, sys.stderr
# The ORIGINAL std streams are bound here, before the program runs, so the
# settlement flush can drain bytes a program wrote through them without an
# explicit flush. The bootstrap only replaces `sys.stdout`/`sys.stderr` with
# the `_LogStream`; `sys.__stdout__`/`sys.__stderr__` (and C-ext stdio
# layered on the same fds) are untouched, and their block-buffered bytes are
# lost when the host SIGTERMs the child right after the done frame β the
# default SIGTERM disposition terminates without interpreter finalization.
# Binding the names here (before the program) makes them immune to a
# `sys.__stdout__ = boom` rebind in model code; `None` under `-S`-style
# redirects is guarded at flush time.
# Bind the FLUSH METHODS, not the stream objects: the settlement flush
# loop iterates callables, and a bare TextIOWrapper object is not callable β
# invoking it would raise TypeError and be swallowed by the loop's except,
# silently disabling the drain. A bound method captures its stream at
# binding time, so a later `sys.__stdout__ = boom` rebind cannot redirect
# it; `None` (stream absent) is guarded at flush time.
_stdout_orig = sys.__stdout__.flush if sys.__stdout__ is not None else None
_stderr_orig = sys.__stderr__.flush if sys.__stderr__ is not None else None
# 6. Compile the program as the body of an async function, matching the
# seam contract (`PtcRunRequest.program` is an async-function body: top-level
# `await` and `return` both work, and the returned value is the completion).
# AST-splicing the parsed body into an `async def` keeps every statement's
# original line number, so a traceback points at the model's own source.
ns: dict[str, Any] = {
"__name__": "__main__",
"__builtins__": __builtins__,
**namespaces,
}
# Read the enforcement callable and its budget into this frame's locals
# BEFORE the program runs: model code can rebind this module's globals
# (the bootstrap IS ``__main__``), and a frame local is not a module
# attribute, so a later ``__main__._DIE_IF_CPU_EXHAUSTED = ...`` cannot
# change which callable the post-check below invokes. This defeats the
# one-line rebind, not a determined `sys._getframe` walk; the unforgeable
# bounds are the RLIMIT_CPU hard limit and the host wall clock
# (see _make_cpu_enforcer).
die_if_cpu_exhausted = _DIE_IF_CPU_EXHAUSTED
# The settlement recheck compares against the EFFECTIVE soft CPU limit
# (`cpu_soft`, clamped to any stricter inherited limit above), NOT the
# configured `cpuSeconds`. When the deployment inherited a soft limit below
# the configured value, a program that traps SIGXCPU, burns past the
# inherited soft, and returns inside the soft-to-hard gap must be reported as
# a timeout β checking the configured value would falsely pass it and bypass
# the inherited limit.
cpu_seconds = cpu_soft
# Same capture, same reason, for the failure path and the send that follows
# it. The reporter was a module-global lookup inside the `except` block, so
# ``import __main__; __main__._SAFE_MODEL_TRACEBACK = ...`` put model code
# there with no guard around it; the flush and send were attribute lookups
# on the stream and channel CLASSES, which ``__main__._LogStream.flush_line
# = ...`` rebinds just as easily. All four run AFTER the handler, where a
# throw costs the `done` frame and the host reports a wall-clock timeout
# instead of the model's exception. Binding the callables now fixes what
# runs; what they in turn reach is closed over in _make_failure_reporter.
safe_model_traceback = _SAFE_MODEL_TRACEBACK
flush_out = out_stream.flush_line
flush_err = err_stream.flush_line
# `done` is either a pre-encoded frame STRING (a `_done_with_value` success:
# the completion value was serialized at its validation point, inside the try,
# so a later send never re-walks the live value a mutating daemon thread could
# have changed) or a dict ERROR frame (a rejection or the exception handler,
# which carry no live model value). `send_done` posts whichever form, going
# DIRECTLY through a bound `_encode_json_plain` and a bound `write_encoded` β
# never through `channel.send_sync`, whose body re-resolves `self.write_encoded`
# and `self`'s module-level `_encode_json_plain` at call time.
#
# The program runs as `__main__`, so `import __main__; __main__.ProtocolChannel
# .send_sync = boom` or `__main__._encode_json_plain = boom` would otherwise
# re-resolve the send/encode to a rebranded callable at call time and, when
# that replacement raises, skip the `done` frame and downgrade a settled
# verdict to a host-side worker-exit (the binding-all-names regression test
# pins this). Same reason `flush_out`/`flush_err`/`safe_model_traceback` are
# bound above.
encode_plain_bound = _encode_json_plain
write_encoded_bound = channel.write_encoded
# The completion-frame builder is bound into a LOCAL here, before the
# program runs: `done = _done_with_value(...)` below sits after the program
# (which is `__main__`) may have rebound `__main__._done_with_value`, so a
# module-global lookup at call time would let a one-line rebind rewrite a
# legitimate success into an `exception`. Binding it (with its own def-time
# default-captured `_check_done_value`/`_encode_json_plain`) makes the entry
# name immune.
done_with_value_bound = _done_with_value
# The fallback primitives are bound into LOCALS here, before the program
# runs, so `send_done`'s except arm does not read module globals at call
# time. This bootstrap is `__main__`, so `__main__._os_write = boom` (or
# `__main__._FALLBACK_DONE_FRAME`, `__main__._memoryview`) would otherwise
# rebind exactly the names the fallback reads, reopening the single-line-
# rebind hole the fallback exists to close.
_os_write_local = _os_write
_memoryview_local = _memoryview
_fallback_frame_local = _FALLBACK_DONE_FRAME
def send_done(payload: dict[str, Any] | str) -> None:
try:
if _isinstance(payload, _str):
write_encoded_bound(payload)
else:
write_encoded_bound(encode_plain_bound(payload))
except: # noqa: BLE001, E722 -- a rebind must not cost the done frame; bare except avoids naming BaseException
# `encode_plain_bound`/`write_encoded_bound` are bound callables, but
# their BODIES still resolve transitive module globals at call time β
# `_encode_json_plain` reaches `_dump_scalar`/`_dump_string`/`json.dumps`,
# `write_encoded` reaches `os.write` (via the `os` module). This
# bootstrap is `__main__`, so `__main__._dump_scalar = boom` (or
# `__main__.os = ...`) makes the error-frame encode/write throw AFTER
# the `except` block, which would drop the `done` frame and downgrade a
# settled `exception` verdict to a host-side `worker-exit`. Write a fixed
# literal done frame with the LOCALLY-BOUND `_os_write_local` and
# `_memoryview_local` (captured before the program runs, so a one-line
# rebind of the module global cannot change them) so the host still gets
# a verdict. The literal is JSON-valid and newline-terminated; the lock
# is the channel's, so the write is serialized against any concurrent
# writer.
with channel._write_lock:
view = _memoryview_local(_fallback_frame_local)
while view:
view = view[_os_write_local(channel._fd, view):]
max_value_bytes = int(boot["maxValueBytes"])
done: dict[str, Any] | str
try:
# filename="<model>" keeps the source label consistent with the later
# compile(wrapped, "<model>", ...) and the runtime traceback filtering
# (safe_model_traceback drops frames whose filename is not "<model>");
# the default "<unknown>" would leak a different label into model-visible
# syntax diagnostics.
module = ast.parse(program, filename="<model>")
wrapper = ast.AsyncFunctionDef(
name="__dsh_main__",
args=ast.arguments(
posonlyargs=[], args=[], vararg=None,
kwonlyargs=[], kw_defaults=[], kwarg=None, defaults=[],
),
body=module.body or [ast.Pass()],
decorator_list=[],
returns=None,
)
# Anchor the synthetic wrapper on the first real statement (or line 1 for
# an empty program) so fix_missing_locations does not stamp it at 0.
anchor = module.body[0] if module.body else ast.parse("pass").body[0]
ast.copy_location(wrapper, anchor)
wrapped = ast.Module(body=[wrapper], type_ignores=[])
ast.fix_missing_locations(wrapped)
# `dont_inherit=True` stops this module's `from __future__ import
# annotations` (line 14) from leaking into the program's compile: PEP 563
# would otherwise stringify the program's type annotations, changing the
# semantics of a legal program that reads `f.__annotations__` at runtime.
code = compile(wrapped, "<model>", "exec", dont_inherit=True)
exec(code, ns) # noqa: S102 -- defines __dsh_main__; executing model code is the point
value = await ns["__dsh_main__"]()
die_if_cpu_exhausted(cpu_seconds)
# Flush the log buffers BEFORE metering and framing the completion value.
# `_done_with_value` materializes the value's escaped JSON form to meter
# it and then pre-encodes the admitted value into its frame (see its
# docstring for the TOCTOU rationale) β several copies of a near-budget
# value live at once (see OUTPUT_BUDGET_WORST_CASE_ADDRESS_SPACE_MULTIPLE).
# Any unflushed log pending would add its own bytes to that peak, so a
# `maxLogBytes` and a `maxValueBytes` each admitted alone by the load gate
# could together breach RLIMIT_AS. Flushing first frees the log pending so
# the value frame's peak stands alone against the address space.
flush_out()
flush_err()
done = done_with_value_bound(value, max_value_bytes)
except _BaseException as exc: # noqa: BLE001 -- report every failure to host; `_BaseException` is a pre-program local, not a rebindable module global
done = {
"type": "done",
"error": {
"kind": "exception",
# Cap the diagnostic BEFORE it crosses the wire: a program can
# raise with a gigabytes-long message, and formatting/sending
# it whole would allocate on both sides before the host's own
# cap runs. Byte-cap at maxValueBytes with the host's marker
# text so the truncated diagnostic reads identically wherever
# the cap was applied. The rendering is wrapped because the
# `done` send below sits outside this handler: a throw while
# formatting would skip it and strand the host on fd 3 until
# maxWallMs (see _make_failure_reporter).
"message": safe_model_traceback(exc, max_value_bytes),
},
}
# Flush any print output not terminated by a newline (a traceback always
# ends in one, but `print(x, end="")` or a bare write may not), so the
# final partial line is not silently dropped. The success path already
# flushed before framing the value; this is an idempotent no-op there and
# the flush the exception path needs.
#
# Guarded because `done` is ALREADY DECIDED here: on the exception path the
# handler above built it, and a flush that raises (its join/encode under
# memory pressure, after the program left a near-maxLogBytes pending and then
# allocated toward RLIMIT_AS) would skip `send_done` and downgrade a run the
# child already classified as `exception` into a host-side `worker-exit`.
# Losing the log tail is the lesser outcome, and the marker the ledger
# already pushed still reports the truncation. Same rule as
# `_make_failure_reporter`: a settled verdict must not be swallowed by the
# reporting that follows it.
for _flush in (flush_out, flush_err, _stdout_orig, _stderr_orig):
if _flush is None:
continue
try:
_flush()
except _BaseException: # noqa: BLE001 -- swallow ONLY the log tail; `done` must reach the host; `_BaseException` is a pre-program local
pass
reply_task.cancel()
send_done(done)
async def _pump_replies(
channel: ProtocolChannel,
pending: dict[int, tuple[asyncio.AbstractEventLoop, asyncio.Future[Any]]],
pending_lock: "threading.Lock",
# The frame reader is a bound method captured by _run BEFORE the program
# runs (see the create_task site), so a rebind of the class attribute cannot
# redirect it.
_read_frame: Any,
# Bound as DEFAULT ARGUMENTS so they are captured at def/import time, before
# ANY model code runs. This bootstrap IS `__main__`, so `__main__.RuntimeError
# = ...` (or `__main__._BindingRejection`, `__main__.str`, `__main__.bool`)
# as a program top-level statement would otherwise rebind the module globals
# these clauses resolve at runtime. A body-local `X = X` binding is too late:
# `_run` reaches `await __dsh_main__` (whose top-level statements run first)
# with no suspension point after `create_task`, so the model's rebind executes
# before the pump body. Defaults are evaluated in the enclosing scope at def
# time, truly before the program.
_RuntimeError: Any = RuntimeError,
_BindingRejection: Any = _BindingRejection,
_str: Any = str,
_bool: Any = bool,
) -> None:
"""Background task: read reply frames and settle pending futures.
Cancelled after ``done`` is posted. Unknown ids and post-settlement replies
are ignored (mirrors the worker backend's hostile-peer stance, though here
the host is the trusted side; the guards defend against races).
A pending Future may belong to a loop other than this pump's β the model can
call a binding from a thread running its own loop (``asyncio.run(tools.x())``).
``asyncio.Future`` is not thread-safe, so the completion is scheduled on the
Future's OWN loop via ``call_soon_threadsafe`` rather than mutated here; a
direct ``set_result`` would never wake the waiting loop and the call would
hang to the wall clock. The ``pop`` shares ``pending_lock`` with ``dispatch``
so a reply cannot race the claim that registers its id.
"""
def complete(fut: asyncio.Future[Any], ok: bool, value: Any, message: Any) -> None:
# Runs on the Future's own loop. `done()` re-checked here because
# cancellation or a duplicate reply may have settled it between the pop
# and this callback.
if fut.done():
return
if ok:
fut.set_result(value)
else:
fut.set_exception(_BindingRejection(_str(message)))
while True:
frame = await _read_frame()
if frame is None:
return
if frame.get("type") != "reply":
continue
with pending_lock:
entry = pending.pop(frame.get("id"), None)
if entry is None:
continue
loop, fut = entry
ok = _bool(frame.get("ok"))
value = frame.get("value")
message = frame.get("message")
try:
loop.call_soon_threadsafe(complete, fut, ok, value, message)
except _RuntimeError: # `_RuntimeError` is a pre-program local, not a rebindable module global
# The Future's loop has already closed β the thread that ran
# `asyncio.run(tools.x(...))` finished (its coroutine was cancelled
# or it exited) before this reply arrived, so nothing awaits the
# Future and the reply is moot. Drop it; scheduling onto a closed
# loop raises RuntimeError, and letting that escape would kill the
# pump and strand every later reply β the exact failure class this
# cross-loop delivery exists to prevent. An abandoned call's pending
# entry is not leaked: it is popped here when its reply arrives
# (dispatch's cancellation does not remove it), so stranded entries
# are bounded by the number of calls THIS run itself issued.
continue
def _encode_json_plain(value: Any) -> str:
"""Encode JSON-plain data iteratively, byte-identical to compact ``json.dumps``.
``json.dumps`` recurses one Python frame per nesting level and raises
``RecursionError`` a few thousand levels deep, but the seam's
``PtcJsonValue`` has no depth limit β a valid deeply nested completion or
call argument below the byte budget must cross intact (the host uses the
same iterative idiom in ``protocol.ts``). Accepts what the callers already
validated or constructed: ``None``/``bool``/``int``/finite ``float``/
``str``, exact ``list``/``tuple``, and exact ``dict`` with ``str`` keys.
Scalar encoding delegates to ``json.dumps`` (string escaping, float repr)
so the bytes match; non-finite floats still raise ``ValueError`` exactly
like ``allow_nan=False``.
Containers are classified by EXACT type and traversed through the unbound
built-in methods rather than the instance's own: a ``dict``/``list``
subclass can override ``items``, ``keys``, ``__iter__``, ``__len__``, or
``__getitem__``, and the validators only see the container it subclasses,
so an instance-method call here could emit different data than the walk
that metered and approved it. ``_check_done_value`` and
``_lossless_json_violation`` reject subclasses outright, so this path only
ever sees exact containers; classifying on exact type keeps that agreement
checkable at one glance instead of resting on the caller.
"""
# O(DEPTH) auxiliary space, not O(width). A container pushes ONE cursor frame
# that pulls its children one at a time and writes each into the shared buffer,
# rather than one stack entry (plus a separator marker) per child: a flat
# `[0] * 6_000_000` encodes to ~12 MB but per-element frames are ~400 MB β an
# RLIMIT_AS death on a value `_check_done_value` already admitted (it walks by
# depth as well). The output string is the only width-proportional
# allocation, and its size the caller metered within budget. `io.StringIO`
# accumulates without the intermediate `"".join(chunks)` second copy. A cursor
# frame is [kind, iterator, wrote_any]; a visit frame is (VISIT, value).
buffer = io.StringIO()
exhausted = object()
visit, list_cursor, dict_cursor = 0, 1, 2
stack: list[Any] = [(visit, value)]
while stack:
frame = stack.pop()
kind = frame[0]
if kind == list_cursor:
iterator, wrote_any = frame[1], frame[2]
child = next(iterator, exhausted)
if child is exhausted:
buffer.write("]")
continue
if wrote_any:
buffer.write(",")
else:
frame[2] = True
stack.append(frame)
stack.append((visit, child))
continue
if kind == dict_cursor:
iterator, wrote_any = frame[1], frame[2]
entry = next(iterator, exhausted)
if entry is exhausted:
buffer.write("}")
continue
key, item = entry
if wrote_any:
buffer.write(",")
else:
frame[2] = True
buffer.write(_dump_scalar(key))
buffer.write(":")
stack.append(frame)
stack.append((visit, item))
continue
current = frame[1]
current_type = type(current)
if current_type is list or current_type is tuple:
buffer.write("[")
stack.append([list_cursor, iter(current), False])
elif current_type is dict:
buffer.write("{")
stack.append([dict_cursor, iter(dict.items(current)), False])
else:
buffer.write(_dump_scalar(current))
return buffer.getvalue()
def _dump_scalar(value: Any) -> str:
"""One scalar as compact JSON, byte-compatible with the host's encoder.
``ensure_ascii=False`` keeps non-ASCII text as raw UTF-8 β the default
backslash-u escaping would make the child count ``"Γ©"`` as 8 bytes where
the host meter (and the worker backend) count its UTF-8 JSON form as 4,
splitting the budget the two sides are supposed to share. Strings route
through :func:`_dump_string`, which restores the escaping for the one class
of character UTF-8 cannot hold. Floats route through :func:`_dump_float`
because CPython's ``repr`` and ECMAScript's Number-to-String disagree on
spelling.
Dispatch is on EXACT type, matching the validators: a ``float`` subclass
reaching :func:`_dump_float` would have its overridden ``__repr__`` read as
the number's digits, so ``F(2.5)`` whose ``__repr__`` returns ``"1.0"``
would serialize as ``1``. ``json.dumps`` then refuses any subclass by
``TypeError`` instead of emitting a value nothing validated; the callers
reject subclasses first, so this is the encoder refusing to be the place a
validation gap turns into corrupted output.
"""
if type(value) is float:
return _dump_float(value)
if type(value) is str:
return _dump_string(value)
if value is None or type(value) is bool or type(value) is int:
return json.dumps(value, ensure_ascii=False, allow_nan=False)
raise TypeError(f"unsupported type ({type(value).__name__})")
# A surrogate code unit, and an adjacent high-low pair. Python stores an astral
# character as ONE code point, so a surrogate reaching these patterns is either
# lone or half of a pair the program spelled out code unit by code unit.
_SURROGATE = re.compile("[\ud800-\udfff]")
_SURROGATE_PAIR = re.compile("[\ud800-\udbff][\udc00-\udfff]")
def _combine_surrogate_pair(match: re.Match[str]) -> str:
"""Fold one spelled-out high-low pair into the astral code point it names."""
high, low = match.group(0)
return chr(0x10000 + ((ord(high) - 0xD800) << 10) + (ord(low) - 0xDC00))
def _dump_string(text: str) -> str:
"""One string as compact JSON, byte-identical to the host's ``JSON.stringify``.
``ensure_ascii=False`` cannot render a surrogate code unit: UTF-8 has no
encoding for one, so the frame write would raise and the run would strand
until the wall clock. JSON carries it as the ASCII escape ``\\ud800``, which
the host's ``JSON.parse`` reads back as the same UTF-16 code unit and its
``JSON.stringify`` re-emits identically β so the shared seam
(``PtcJsonValue``, ``snapshotJsonValue``, the worker backend) keeps a
lone-surrogate string instead of failing the value. An adjacent high-low
pair is folded into its astral code point FIRST: the host holds strings as
UTF-16, where those two code units and the single character are the same
string, and the raw 4-byte form is what the host would emit β escaping the
halves separately would charge 12 bytes against a budget the host meters at
4. Every remaining surrogate is lone and becomes six ASCII bytes, matching
the host exactly.
@param text: the string to encode.
@return: its compact JSON form, always UTF-8-encodable.
"""
rendered = json.dumps(text, ensure_ascii=False)
if _SURROGATE.search(rendered) is None:
return rendered
return _SURROGATE.sub(
lambda match: "\\u%04x" % ord(match.group(0)),
_SURROGATE_PAIR.sub(_combine_surrogate_pair, rendered),
)
# How many bytes each byte that needs escaping adds beyond its raw self, as a
# ready-made (byte, surcharge) list so :func:`_json_string_cost` walks no
# branches per pass. ``"`` and ``\\`` take a one-character prefix; the five C0
# controls with a shorthand (``\\b\\f\\n\\r\\t``) likewise; every other C0
# control becomes a six-character ``\\uXXXX``.
_JSON_ESCAPE_SURCHARGES = [
(bytes((byte,)), 1 if byte in b'"\\\b\f\n\r\t' else 5)
for byte in [*range(0x20), ord('"'), ord("\\")]
]
# Per-byte JSON-string serialized cost (the byte itself plus its escape
# surcharge), indexed by byte value. Lets :func:`_cap_message` accumulate the
# serialized cost of a growing prefix in one O(1) step per byte without building
# the escaped form. A non-ASCII byte stays raw (cost 1); a C0 control or ``"``/
# ``\\`` carries its surcharge from :data:`_JSON_ESCAPE_SURCHARGES`.
_JSON_BYTE_COST = [1] * 256
for _escaped_byte, _surcharge in _JSON_ESCAPE_SURCHARGES:
_JSON_BYTE_COST[_escaped_byte[0]] = 1 + _surcharge
def _json_str_cost(text: str) -> int:
"""Byte length of ``text``'s JSON string form, WITHOUT building that form.
The str-side twin of :func:`_json_string_cost`, for the completion-value
meter. Measuring by materializing ``_dump_string(text).encode()`` allocates
the escaped copy plus its encode -- for a NUL-heavy string that is ~6x the
original each, so metering a value the budget would have REJECTED could
itself breach ``RLIMIT_AS`` and report ``exception`` where the contract
promises ``output-limit``.
The common case encodes once (~1x, well inside the load gate's envelope) and
counts escapes with the same C-level passes :func:`_json_string_cost` uses.
A string carrying surrogate code units has no UTF-8 form at all, so it takes
the exact path :func:`_dump_string` defines: fold each spelled-out high-low
pair into its astral character first (the host meters that as its raw 4-byte
form), then charge six ASCII bytes for every surviving lone surrogate and
count the rest from its encodable remainder.
@param text: the string to measure.
@return: the byte length of its JSON string form, quotes included.
"""
try:
return _json_string_cost(text.encode("utf-8"))
except UnicodeEncodeError:
pass
folded = _SURROGATE_PAIR.sub(_combine_surrogate_pair, text)
# Remove the lone surrogates first, then count them as the length
# difference: `_SURROGATE.findall(folded)` materialized one single-character
# string PER surrogate, so a surrogate-dense value near the budget
# (millions of lone surrogates, each serializing to six bytes) allocated
# millions of objects before the meter returned -- an RLIMIT_AS death
# surfacing as `exception` instead of the promised `output-limit`. After
# pair-combining, every remaining surrogate is lone and exactly one code
# point, so the removed length is the count, and the `without` string is
# needed for the meter anyway.
without = _SURROGATE.sub("", folded)
lone = len(folded) - len(without)
# Six ASCII bytes per lone surrogate; the remainder is ordinary text whose
# own quotes are dropped here because the outer call adds them once.
return _json_string_cost(without.encode("utf-8")) + lone * 6
def _json_string_cost(raw: bytes) -> int:
"""UTF-8 byte length of one string's JSON form, WITHOUT building that form.
Used by :class:`LogBuffer` to charge a log entry what it will actually cost
on the wire. Building ``json.dumps(text)`` to measure it would allocate a
second copy up to six times the original β the very allocation the ledger's
cheap pre-check exists to avoid, and enough to breach ``RLIMIT_AS`` on a
large control-heavy line. Counts exactly what :func:`_dump_scalar`'s
``ensure_ascii=False`` output holds: the two quotes, each escaped byte's
surcharge from :data:`_JSON_ESCAPE_SURCHARGES`, and the raw bytes themselves
(non-ASCII stays raw, so its UTF-8 length already counts). Uses a fixed
number of C-level ``count`` passes β allocating nothing, unlike a
``translate`` filter β because the caller admits up to ~4x the remaining
budget of bytes here and a per-byte Python loop over it would cost more than
the encode being avoided.
@param raw: the entry's UTF-8 bytes.
@return: the byte length of its JSON string form, quotes included.
"""
extra = 0
for byte, surcharge in _JSON_ESCAPE_SURCHARGES:
extra += raw.count(byte) * surcharge
return len(raw) + 2 + extra
def _dump_float(value: float) -> str:
"""One finite float in ECMAScript ``Number::toString`` spelling.
CPython's ``repr`` and the host's ``String(number)`` name the same double
differently: ``1.0`` is ``"1.0"`` here but ``"1"`` there, ``1e-07`` pads the
exponent the host writes as ``1e-7``, and ``1e+21``/``2**60`` differ again.
Since the child meters the completion value against ``maxValueBytes`` and
the host re-meters the frame it parses, any spelling difference splits the
shared budget: ``return 1.0`` under ``maxValueBytes: 1`` used to be reported
as ``output-limit`` by the child while the host would have counted the
one-byte ``1`` it actually receives. Both sides also emit these bytes (the
child through :func:`_encode_json_plain`, the host through
``encodeJsonPlain``), so the fix has to be in the shared speller, not in the
meter.
Implements ECMA-262 ``Number::toString`` radix 10 directly: ``repr``
already yields the shortest round-tripping decimal digits, and ``Decimal``
splits them into the significand ``s`` (``digits``, ``k`` of them) and
decimal exponent ``n`` the spec's cases select on. The integral values above
the JS safe range take the host's BigInt branch, whose exact digits differ
from the shortest-round-trip form (``2**60`` prints ``...846976``, not
``...847000``).
"""
if value != value or value in (float("inf"), float("-inf")):
# json.dumps(allow_nan=False) raises the same way; the callers reject
# non-finite floats before metering, so this is unreachable defense.
raise ValueError("Out of range float values are not JSON compliant")
if value == 0.0:
# Covers -0.0 too; callers reject it as non-lossless before this point.
return "0"
if value < 0:
return "-" + _dump_float(-value)
if value.is_integer() and value > float(2**53 - 1):
# The host's BigInt branch: exact digits, not shortest-round-trip.
return str(int(value))
parts = Decimal(repr(value)).normalize(context=_FLOAT_CONTEXT).as_tuple()
digits = "".join(str(digit) for digit in parts.digits)
k = len(digits)
n = parts.exponent + k
if k <= n <= 21:
return digits + "0" * (n - k)
if 0 < n <= 21:
return digits[:n] + "." + digits[n:]
if -6 < n <= 0:
return "0." + "0" * -n + digits
exponent = ("+" if n - 1 >= 0 else "-") + str(abs(n - 1))
return (digits if k == 1 else digits[0] + "." + digits[1:]) + "e" + exponent
def _check_done_value(value: Any, max_bytes: int):
"""Meter a completion value's JSON byte size AND validate its lossless-JSON
shape in one bounded post-order walk; return ``None`` when it passes.
Folds what was formerly a losslessness walk followed by a separate byte
meter into one pass. Running the losslessness walk first materialized one
traversal tuple per element before any size cap: ``return [0] * 2000000``
under ``maxValueBytes: 64`` allocated millions of frames (an RLIMIT_AS
death) before the meter could reject it. Folding the byte bound into the
walk rejects over-budget BEFORE enqueuing a container's children β every
element is at least one JSON byte β so the walk stays O(cap). Same
JS-double-exact integer boundary, cycle detection (a leave marker pops each
container off ``on_path``), and type rejections as
:func:`_lossless_json_violation`, and the same byte accounting as
:func:`_encode_json_plain`. :func:`_lossless_json_violation` stays for the
binding-argument path, which carries no size cap.
EVERY type here is matched EXACTLY, containers and scalars alike, so a
subclass is rejected as an unsupported type rather than admitted by
``isinstance``. A subclass can override the operators and methods this walk
and the encoder call, and they need not agree: a populated ``dict``
subclass whose ``items()`` returns ``[]`` would meter as ``{}``; a ``float``
subclass overriding ``__repr__`` passes the non-finite and negative-zero
checks by its real value but serializes as whatever the override says, since
:func:`_dump_float` reads ``repr``; an ``int`` subclass overriding ``__gt__``
and ``__lt__`` slips past the JS-safe-range bound while ``json.dumps``
emits its true C-level digits, so ``2**53 + 1`` reaches the host as
``...992``; a ``str`` subclass overriding ``__len__`` returns 0 from the
pre-encode lower bound and admits an arbitrarily large string. In each case
the value the host receives differs from the one this walk approved. The
worker backend rejects the equivalent shapes by prototype identity and
``typeof`` (``hasPlainObjectPrototype`` in ``worker-json.ts``); a ``bool``
is checked before ``int`` because it is an ``int`` subclass that IS
lossless JSON.
Returns ``("invalid-output", message)`` for a non-lossless value,
``("output-limit", message)`` once the size crosses ``max_bytes``, or
``None`` when the value is lossless JSON within budget.
Metering and validation interleave in this single traversal: each member is
costed the moment it is visited, and it is rejected the moment it trips
either check. A value that holds BOTH an over-budget member and an
invalid-typed member therefore resolves to whichever tripped FIRST in
visit order β both are rejects, and neither kind claims priority over the
other, so that first-trip order is not part of the seam contract; the
host side independently re-measures the value it receives.
"""
js_safe = 2**53 - 1
def invalid(reason: str):
return ("invalid-output", f"program completion must be lossless JSON ({reason})")
over_budget = ("output-limit", f"completion value exceeded {max_bytes} bytes")
total = 0
on_path: set[int] = set()
# The walk uses O(DEPTH) space, not O(width). A container pushes ONE cursor
# frame that pulls its children one at a time, rather than one traversal
# frame per child: a flat `[0] * 6_000_000` serializes to ~12 MB (well within
# a modest budget) but one tuple per element is ~380 MB β an RLIMIT_AS death
# on a value the byte meter would admit, the very inversion this meter exists
# to prevent. A cursor frame is (kind, container, iterator); a visit frame is
# (VISIT, value, None). The upfront structural bound still rejects a wide
# forgery before any iteration begins.
exhausted = object()
visit, list_cursor, dict_cursor = 0, 1, 2
stack: list[tuple[int, Any, Any, Any]] = [(visit, value, None, None)]
while stack:
frame = stack.pop()
kind = frame[0]
if kind == list_cursor:
container, iterator = frame[1], frame[2]
child = next(iterator, exhausted)
if child is exhausted:
on_path.discard(id(container))
continue
# Resume this cursor after the child is fully walked; the child goes
# on top so it is visited next (order does not affect the byte total).
stack.append(frame)
stack.append((visit, child, None, None))
continue
if kind == dict_cursor:
container, iterator, seen = frame[1], frame[2], frame[3]
entry = next(iterator, exhausted)
if entry is exhausted:
on_path.discard(id(container))
continue
key, item = entry
# Only an EXACT str key survives: bool and int coerce or raise, and a
# str SUBCLASS can override the ``__len__`` the bound below reads while
# the encoder emits its real characters.
if type(key) is not str:
return invalid(f"non-string dict key ({type(key).__name__})")
# The key's JSON form folds a spelled-out surrogate pair into its
# astral code point (`_dump_string`), so two DIFFERENT Python keys β
# the two code units and the single character β encode to the same
# JSON member, and the host's JSON.parse silently drops one of them.
# That is a lossless-JSON violation, so the collision is rejected
# here, before any encoding.
combined_key = _SURROGATE_PAIR.sub(_combine_surrogate_pair, key)
if combined_key in seen:
return invalid("duplicate dict key after surrogate-pair combining")
seen.add(combined_key)
# The same string lower bound, before escaping the key.
if total + len(key) + 3 > max_bytes:
return over_budget
# Same counting rule as the string branch: a control-heavy KEY
# expands just as far, and `_dump_scalar` on a str is `_dump_string`.
total += _json_str_cost(key) + 1
if total > max_bytes:
return over_budget
stack.append(frame)
stack.append((visit, item, None, None))
continue
current = frame[1]
if current is None or type(current) is bool:
total += len(_dump_scalar(current).encode("utf-8"))
elif type(current) is str:
# Lower-bound BEFORE materializing the escaped form: every character
# is at least one UTF-8 byte plus the two quotes, so a huge or
# control-heavy string (whose escaped copy expands severalfold) is
# rejected without allocating that copy.
if total + len(current) + 2 > max_bytes:
return over_budget
# A lone surrogate has no UTF-8 form but a lossless JSON one β the
# ASCII ``\uXXXX`` escape :func:`_dump_string` emits β so it is
# metered, not rejected, matching the shared seam. Metered by
# COUNTING, not by building the escaped form: that copy plus its
# encode is ~6x the original for a control-heavy string, so measuring
# a value the budget rejects could breach RLIMIT_AS and surface as
# `exception` instead of the promised `output-limit`.
total += _json_str_cost(current)
elif type(current) is int:
# The canonical boundary accepts every JS-double-exact value: an int
# outside +-2**53-1 is fine IFF the double round-trip is exact.
if current > js_safe or current < -js_safe:
try:
exact = int(float(current)) == current
except OverflowError:
exact = False
if not exact:
return invalid("integer not exactly representable as a JavaScript number")
total += len(_dump_scalar(current).encode("utf-8"))
elif type(current) is float:
if current != current or current in (float("inf"), float("-inf")):
return invalid("non-finite float")
# JSON turns -0.0 into a sign the host parses back to JS -0; the
# canonical boundary rejects it, so this side must too.
if current == 0.0 and math.copysign(1.0, current) < 0:
return invalid("negative zero")
total += len(_dump_scalar(current).encode("utf-8"))
elif type(current) is list:
if id(current) in on_path:
return invalid("circular reference")
count = len(current)
total += 2 + (count - 1 if count > 1 else 0)
# Reject over-budget BEFORE iterating: every element serializes to at
# least one byte, so a wide flat forgery fails here without pulling a
# single child.
if total + count > max_bytes:
return over_budget
on_path.add(id(current))
stack.append((list_cursor, current, iter(current), None))
elif type(current) is dict:
if id(current) in on_path:
return invalid("circular reference")
# ``len`` without materializing ``current.items()``: that list
# allocates one tuple per member before the bound below could run,
# recreating the spike the bound exists to stop.
count = len(current)
total += 2 + (count - 1 if count > 1 else 0)
# Same pre-iterate bound: each entry contributes a quoted key
# (>= 2 bytes), a colon, and a >= 1-byte value. ``iter`` on the items
# view is O(1); the cursor meters each key as it is pulled.
if total + count * 4 > max_bytes:
return over_budget
on_path.add(id(current))
# The seen-set holds one combined key per member β O(keys), the same
# order as the dict itself β so the surrogate-collision check below
# can detect two keys that fold to one JSON member.
stack.append((dict_cursor, current, iter(current.items()), set()))
else:
# tuple, set, or any other type: not round-trippable JSON.
return invalid(f"unsupported type ({type(current).__name__})")
if total > max_bytes:
return over_budget
return None
def _lossless_json_violation(value: Any) -> str | None:
"""Return why ``value`` is not lossless JSON, or ``None`` when it is.
``json.dumps`` succeeding is NOT proof of losslessness: it coerces a
non-string ``dict`` key to its string form (``{1: "a", "1": "b"}`` collapses
to one key, silently dropping data), emits non-standard ``NaN``/``Infinity``
tokens without ``allow_nan=False``, and accepts integers outside JavaScript's
safe range (``9007199254740993`` becomes ``...992`` once the host parses the
frame into a JS number). Validate the shape up front so a coercive or lossy
value fails as ``invalid-output`` instead of round-tripping to something the
program did not compute. Iterative so deep nesting cannot overflow the stack,
and it tracks the container ancestry on the current path so a cyclic value is
reported at once rather than spinning until the CPU budget. Only JSON-plain
types survive: ``None``/``bool``/JS-safe ``int``/finite ``float``/``str``,
exact ``list``, and exact ``dict`` with ``str`` keys. Every type matches
EXACTLY, containers and scalars alike, for the reason
:func:`_check_done_value` documents: a subclass can override the operators
and methods a traversal calls, so an ``isinstance`` admission here would
approve one shape and let the encoder emit another.
"""
# The canonical boundary accepts every JS-double-exact value: an int
# outside +-2**53-1 is fine IFF the double round-trip is exact (2**53 or
# 2**60 survive; 2**53+1 rounds), matching the worker backend.
js_safe = 2**53 - 1
# Post-order walk with an explicit "leave" marker: a container's id is added
# to `on_path` when entered and removed when left, so a back-edge to an
# ancestor (a cycle) is detected without rejecting a legitimately shared
# acyclic subtree.
on_path: set[int] = set()
# O(DEPTH) auxiliary space, not O(width), for the reason
# :func:`_check_done_value` documents: this walk runs in ``dispatch`` on
# MODEL-CONSTRUCTED binding arguments, which no child-side byte budget
# bounds first (the frame ceiling is the host's, and it applies after this
# returns). Enqueueing one frame per member would let a legitimate
# ``[0] * 6_000_000`` argument -- ~17 MB of JSON -- allocate ~366 MB of
# traversal tuples and die as the program's own MemoryError instead of
# round-tripping. A container therefore pushes ONE cursor frame holding its
# iterator; children are pulled one at a time.
exhausted = object()
visit, container_cursor = 0, 1
# A visit frame is (visit, value, None, None); a cursor frame is
# (cursor, container, iterator, seen-keys-for-dicts).
stack: list[tuple[int, Any, Any, Any]] = [(visit, value, None, None)]
while stack:
kind = stack[-1][0]
if kind == container_cursor:
_, container, iterator, seen = stack[-1]
child = next(iterator, exhausted)
if child is exhausted:
# Leaving the container: it is no longer on the current path, so
# a legitimately shared acyclic subtree is not mistaken for a cycle.
on_path.discard(id(container))
stack.pop()
continue
if type(container) is dict:
# The dict cursor yields (key, value): check the key as it is
# pulled. Only an EXACT str key survives -- int, float, None, and
# tuple keys coerce or raise, and a str subclass can carry
# overrides the encoder does not honor.
key, child = child
if type(key) is not str:
return f"non-string dict key ({type(key).__name__})"
# The key's JSON form folds a spelled-out surrogate pair into its
# astral code point (`_dump_string`), so two DIFFERENT Python
# keys -- the two code units and the single character -- encode
# to the same JSON member, and the host's JSON.parse silently
# drops one of them. A lossless-JSON violation, rejected here
# before any encoding.
combined_key = _SURROGATE_PAIR.sub(_combine_surrogate_pair, key)
if combined_key in seen:
return "duplicate dict key after surrogate-pair combining"
seen.add(combined_key)
stack.append((visit, child, None, None))
continue
_, current, _unused, _unused2 = stack.pop()
if current is None or type(current) is bool:
continue
if type(current) is str:
# Every string is lossless JSON. A lone surrogate has no UTF-8 form,
# but JSON carries the code unit as its ASCII ``\uXXXX`` escape and
# :func:`_dump_string` emits exactly that, so the host receives the
# same code unit the program passed β the same acceptance
# ``PtcJsonValue``, ``snapshotJsonValue``, and the worker backend
# already give it.
continue
if type(current) is int:
if current > js_safe or current < -js_safe:
try:
exact = int(float(current)) == current
except OverflowError:
exact = False
if not exact:
return "integer not exactly representable as a JavaScript number"
continue
if type(current) is float:
if current != current or current in (float("inf"), float("-inf")):
return "non-finite float"
# JSON serialization turns -0.0 into 0 (or "-0.0" text that the
# host parses to JS -0), silently changing the sign bit either
# way; the repository's canonical lossless-JSON boundary and the
# worker backend both reject it, so this side must too.
if current == 0.0 and math.copysign(1.0, current) < 0:
return "negative zero"
continue
if type(current) is list or type(current) is dict:
if id(current) in on_path:
return "circular reference"
on_path.add(id(current))
if type(current) is dict:
# Keys are checked as the cursor pulls each entry, not in a
# separate pass: ``current.values()`` would need a second walk,
# and materializing ``items()`` up front allocates one tuple per
# member -- the very spike the cursor removes. The seen-set holds
# one combined key per member -- O(keys), the same order as the
# dict itself -- for the surrogate-collision check.
stack.append((container_cursor, current, iter(current.items()), set()))
else:
stack.append((container_cursor, current, iter(current), None))
continue
return f"unsupported type ({type(current).__name__})"
return None
def _make_cpu_enforcer() -> Any:
"""Build the CPU post-check over closure-held primitives.
This bootstrap IS ``__main__``, so model code can reach every one of its
module globals: ``import __main__; __main__._X = ...`` rebinds the name the
enforcement would otherwise read at call time, which a plain module-level
function plus module-level captures made a one-line defeat. The primitives
therefore live in this factory's locals, which become closure cells of the
returned function, and :func:`_run` binds the returned function into a
local of its own frame BEFORE executing the program, so no assignment to
``__main__`` changes which callable runs or what it calls. Capture happens
at import time, before model code runs, so the captured
``resource.getrusage``/``signal.signal``/``os.kill`` are the real builtins.
This raises the cost of defeating the check; it does not make it
unreachable, and nothing in-process could. A cell is writable through
``fn.__closure__[i].cell_contents``, and ``sys._getframe`` walks to
:func:`_run`'s frame and reads its locals, so a program determined to
tamper still can β consistent with this backend's documented posture, where
the in-process interpreter is containment rather than a security boundary
(Β§Trust posture in the PTC mode Agent Note). The bounds that model code cannot
forge are outside the interpreter: the RLIMIT_CPU HARD limit at
``cpuSeconds + 1``, whose SIGKILL is undeliverable to a handler and
unraisable by a process that cannot raise its own hard limit, and the
host's wall-clock ceiling. This check exists to convert the two cases those
miss β a program that traps SIGXCPU and settles inside the soft-to-hard
gap, and a program that spends the budget in DESCENDANTS the kernel never
charged to this process β from a reported SUCCESS into the same `timeout`
an untrapped program gets.
@returns The one-argument enforcement callable, taking `cpuSeconds`.
"""
getrusage = resource.getrusage
rusage_self = resource.RUSAGE_SELF
rusage_children = resource.RUSAGE_CHILDREN
set_signal = signal.signal
sig_dfl = signal.SIG_DFL
sigxcpu = signal.SIGXCPU
kill = os.kill
getpid = os.getpid
# SIGXCPU unmasking primitives for the re-raise below: a program can mask
# the signal and return past the soft limit, so the re-delivered signal
# must be unblocked first. Captured here (import time) so a rebind cannot
# defeat them. The ``getattr``/``None`` guard is defensive against a
# stripped CPython build (the host refuses win32 at construction, so every
# platform this backend actually starts on has ``pthread_sigmask``).
pthread_sigmask = getattr(signal, "pthread_sigmask", None)
sig_unblock = getattr(signal, "SIG_UNBLOCK", None)
def die_if_cpu_exhausted(cpu_seconds: int) -> None:
"""Die by re-delivered SIGXCPU when the CPU budget is already spent.
Two cases reach here as a would-be SUCCESS. A model program can trap
SIGXCPU and return during the one-second soft-to-hard gap. And
``RLIMIT_CPU`` is PER-PROCESS, inherited fresh by every child, so a
program calling ``subprocess`` or ``os.fork`` multiplies the run's CPU
budget by the number of descendants it starts: measured with
``cpuSeconds: 1``, two sequential busy children burned 2.0
CPU-seconds and the parent, which had accrued almost no CPU of its own
while blocked in ``subprocess.wait``, still returned a completion.
The meter is therefore ``RUSAGE_SELF + RUSAGE_CHILDREN``, the kernel's
own aggregate, which accumulates the CPU of every REAPED descendant
(grandchildren included, verified).
``getrusage`` is the kernel's own meter (unforgeable from model code),
and dying by SIGXCPU with the default disposition restored gives the
host the same kernel-authoritative close signal as the untrapped soft
limit β classified as `timeout`, after which the host's process-group
SIGTERM/SIGKILL teardown reaches any surviving descendants. Runs AFTER
the model program settled, so a program can re-trap SIGXCPU between
this SIG_DFL and the kill only by running more code, which it no longer
does. A program that tampers with this callable instead (see
:func:`_make_cpu_enforcer` on why in-process state cannot be hidden)
buys at most the remaining soft-to-hard gap: one more CPU second, after
which the hard limit's SIGKILL lands with no handler possible.
Checking at settle time rather than sampling mid-run is deliberate:
both mid-run designs perturb the run they measure. A sampling thread
cost 72 MiB of virtual address space in the child (8 MiB stack plus a
64 MiB glibc per-thread malloc arena reservation; measured 30.23 MiB of
mappings without it against 102.37 MiB with it), and ``RLIMIT_AS``
counts reserved space, so it silently shrank every run's
`addressSpaceMb`. A ``SIGALRM`` interval timer costs no mappings but
makes the program's own syscalls return short under PEP 475 β measured
a 64 MiB ``os.write`` returning 65536 β which corrupts fd-3 framing.
The cost of checking only at settle time is that a descendant's CPU is
detected after it is spent, not while it runs; the host's wall-clock
ceiling bounds that interval, and a program that never reaps its child
is bounded by the wall clock alone, since ``RUSAGE_CHILDREN`` counts
only reaped descendants (verified: a still-running child contributes
0.0).
@param cpu_seconds The `cpuSeconds` budget the soft RLIMIT_CPU used.
"""
own = getrusage(rusage_self)
kids = getrusage(rusage_children)
spent = own.ru_utime + own.ru_stime + kids.ru_utime + kids.ru_stime
if spent >= cpu_seconds:
# A program can mask SIGXCPU (``pthread_sigmask(SIG_BLOCK, ...)``),
# burn past the soft limit, and return during the soft-to-hard gap;
# the re-delivered SIGXCPU below would then stay PENDING and the
# child would exit normally with a success result. Restore the
# default disposition BEFORE unblocking: a program that installed a
# custom handler AND masked the signal has that pending handler run
# the moment the signal is unblocked (CPython delivers it at the next
# eval-breaker checkpoint in model code), and it could re-mask or
# raise β so the disposition must already be SIG_DFL when the signal
# is released. With SIG_DFL restored first, the pending signal kills
# the process inside the kernel with no bytecode window; the ``kill``
# below is the fallback for the never-pending case. ``pthread_sigmask``
# is ``None``-guarded defensively (every platform this backend starts
# on has it; the host refuses win32 at construction).
set_signal(sigxcpu, sig_dfl)
if pthread_sigmask is not None:
pthread_sigmask(sig_unblock, (sigxcpu,))
kill(getpid(), sigxcpu)
return die_if_cpu_exhausted
_DIE_IF_CPU_EXHAUSTED = _make_cpu_enforcer()
_TRUNCATION_MARKER = "β¦ [truncated]"
# The marker's own UTF-8 size, reserved out of the cap rather than added on top
# of it. Byte-identical to the host's TRUNCATION_MARKER_BYTES; the ellipsis is
# three bytes, so this is 15, not the string's 13 characters.
_TRUNCATION_MARKER_BYTES = len(_TRUNCATION_MARKER.encode("utf-8"))
def _cap_message(message: str, max_bytes: int) -> str:
"""Cap a diagnostic by its SERIALIZED cost, appending the host's marker.
Metered by the JSON-string cost the ``done`` frame will actually carry, not
by raw UTF-8 length: the message crosses fd 3 inside a JSON frame where
control characters escape up to sixfold (a NUL is one raw byte but six as
``\\u0000``), so a raw-length cap of ``maxValueBytes`` could serialize to
roughly six times that and breach the 64 MiB frame parse cap β the silent
``worker-exit`` inversion the load-time cap check exists to prevent, and a
several-hundred-MiB escape allocation besides. The seam's load bound admits
``maxValueBytes`` up to ``parse-cap - envelope`` on the premise that both the
completion value and the diagnostic are metered in serialized bytes, so this
honors that premise for the diagnostic.
Encoded with ``errors="replace"`` first: a model exception message can
contain an unpaired surrogate (``raise Exception("\\ud800")``), and a strict
encode would throw while BUILDING the failure frame β the run would then
strand until the wall clock instead of reporting the exception. The marker's
serialized cost comes OUT of ``max_bytes``, so the returned string's own
frame form honors the cap; the host meters the same field again on arrival.
A ``max_bytes`` below the marker's cost yields the marker alone.
This is the PRODUCING-side cap. The host's receive-side ``capMessage``
(``src/index.ts``) bills the same field by RAW bytes instead, because its
output goes into ``PtcRunResult.error.message`` and never re-crosses a
frame-bounded channel β see that function's JSDoc for the split.
"""
raw = message.encode("utf-8", errors="replace")
if _json_string_cost(raw) <= max_bytes:
return raw.decode("utf-8")
# Truncating: the result is `prefix + marker`, whose serialized cost is
# `2 (quotes) + sum(prefix byte costs) + marker cost`. The marker is
# escape-free, so its cost is its UTF-8 length. Reserve that and the quotes,
# then take the longest raw prefix whose accumulated per-byte cost fits.
# `_JSON_BYTE_COST` is per-byte and additive, so the scan is exact and walks
# at most a budget's worth of bytes, allocating nothing (unlike building the
# escaped form). `max(0, ...)` handles a `max_bytes` below the marker's own
# cost, yielding the marker alone.
content_budget = max(0, max_bytes - 2 - _TRUNCATION_MARKER_BYTES)
cost = 0
end = 0
for end in range(len(raw)):
cost += _JSON_BYTE_COST[raw[end]]
if cost > content_budget:
break
else:
end = len(raw)
# Drop a trailing partial UTF-8 sequence the slice may have cut (continuation
# bytes are 0b10xxxxxx); `errors="ignore"` renders the clean prefix.
return raw[:end].decode("utf-8", errors="ignore") + _TRUNCATION_MARKER
# Fixed safety/liveness bound, not a tunable: a model can raise an exception
# with an arbitrarily deep __cause__/__context__ chain, and both the rendering
# walk and format() are linear in chain length. Capping how many links get
# RENDERED keeps traceback formatting from consuming the whole wall budget.
# 100 links is far beyond any legible human traceback.
_MAX_TRACEBACK_CHAIN = 100
# Diagnostic used when rendering the failure itself fails. Built from a fixed
# literal plus the exception CLASS name, never from the exception's own str.
_UNRENDERABLE_DIAGNOSTIC = "<diagnostic rendering failed>"
def _model_traceback(exc: BaseException, max_bytes: int) -> str:
"""Format a model-program failure with only the MODEL's own frames.
Bootstrap frames carry host-absolute paths β meaningless to the model and
unstable across machines, so transcripts pinning them cannot replay. They
appear not only as a leading prefix (the bootstrap's ``exec``/``await``)
but also interleaved and trailing: an uncaught binding rejection re-raised
by ``dispatch`` puts bootstrap frames AFTER the model's, and chained
``__cause__``/``__context__`` exceptions carry their own stacks. Filter
every non-``<model>`` frame across the whole chain rather than trimming a
prefix. A failure with no model frame anywhere (e.g. a SyntaxError raised
by ``compile``) keeps the standard exception-only rendering.
Rendering is bounded to ``_MAX_TRACEBACK_CHAIN`` links, cut on the
``TracebackException`` COPY, and a marker line announces the truncation.
Nothing here touches the live exception: an exception class overriding
``__setattr__`` would run MODEL code from inside the caller's failure
handler, and a throw there costs the ``done`` frame (see
``_safe_model_traceback``). ``TracebackException`` instances hold no such
hooks, so clearing their links runs no model code. The walk is iterative,
so a deep chain cannot overflow the recursion limit.
``from_exception`` still copies the WHOLE live chain, at a higher per-link
cost than building it took. That is bounded by the child's ``RLIMIT_AS``:
the model must materialize every link (exception object plus traceback)
before raising, so a chain long enough for the copy to matter is already
near the address-space cap, and a ``MemoryError`` in the copy lands in the
caller's fallback rather than stranding the run.
"""
te = traceback.TracebackException.from_exception(exc)
# One iterative pass over the copy does both jobs: keep only <model> frames
# on every linked exception and group member, and cut the chain at the cap.
found = False
truncated = False
pending = [(te, 1)]
while pending:
entry, depth = pending.pop()
kept = [f for f in entry.stack if f.filename == "<model>"]
entry.stack = traceback.StackSummary.from_list(kept)
found = found or bool(kept)
# 3.11+ exception groups (a binding failure inside asyncio.TaskGroup)
# carry member stacks under `exceptions`, not the dunder links; a group
# member counts as a link so the cap bounds nesting through both edges.
members = getattr(entry, "exceptions", None) or ()
if depth >= _MAX_TRACEBACK_CHAIN:
if entry.__cause__ is not None or entry.__context__ is not None or members:
truncated = True
entry.__cause__ = None
entry.__context__ = None
if members:
entry.exceptions = None
continue
for linked in (entry.__cause__, entry.__context__):
if linked is not None:
pending.append((linked, depth + 1))
for member in members:
pending.append((member, depth + 1))
def emit():
if found:
yield from te.format()
else:
yield from traceback.format_exception_only(type(exc), exc)
if truncated:
yield f"[dsh-ptc-runtime-python] exception chain truncated at {_MAX_TRACEBACK_CHAIN} links\n"
return _join_bounded(emit(), max_bytes)
def _make_failure_reporter() -> Any:
"""Build the failure-diagnostic renderer over closure-held primitives.
The returned callable renders a model failure diagnostic that cannot itself
raise. The caller sends the ``done`` frame AFTER its ``except BaseException``
block, so anything thrown while rendering the diagnostic skips the send
entirely: the host then blocks on fd 3 until ``maxWallMs`` and reports a
timeout instead of the exception that actually happened. Rendering runs
model code by design (``format()`` reaches ``__str__``, ``__repr__`` and
``__notes__``) and allocates under ``RLIMIT_AS``, so it must be treated as
able to throw.
The fallback names the exception CLASS and a fixed literal β no ``str(exc)``
and no ``format_exception_only``, both of which reach the model's
``__str__``. A ``__name__`` that is not exactly ``str`` (a metaclass
property can return anything, or raise) is discarded rather than
formatted, so no override runs on this path either.
The factory exists for the same reason :func:`_make_cpu_enforcer` does: this
bootstrap IS ``__main__``, so ``import __main__; __main__._X = ...`` rebinds
any module global a call-time lookup would read. On this path a rebind is
worst β the handler's own reporter, and everything the reporter reaches,
would run model code outside any guard, and a throw there costs the ``done``
frame. The traceback formatter, the byte cap and the fallback literal
therefore become closure cells captured at import time, before model code
runs, and :func:`_run` binds the returned callable into a local of its own
frame. A frame local is not a module attribute, so no assignment to
``__main__`` changes which callable runs or what it calls. This defeats the
one-line rebind, not a determined ``sys._getframe`` walk; the unforgeable
bound is the host wall clock.
"""
cap_message = _cap_message
model_traceback = _model_traceback
unrenderable = _UNRENDERABLE_DIAGNOSTIC
# The exception class the guards below catch is bound into a closure cell
# here, at import time, before model code runs. `safe_model_traceback` runs
# AFTER the program (which is `__main__`) may have rebound the module global
# `BaseException`, so `except BaseException` would resolve the rebound class
# and a render-time throw could escape β losing the done frame. A closure cell
# is not reachable by `__main__._X = ...`, so the catch is immune.
_BaseException = BaseException
def safe_model_traceback(exc: BaseException, max_bytes: int) -> str:
try:
return cap_message(model_traceback(exc, max_bytes), max_bytes)
except _BaseException: # noqa: BLE001 -- a throw here would cost the done frame
pass
try:
raw_name = type(exc).__name__
# Slice BEFORE interpolating. A metaclass `__name__` property can
# return an arbitrarily long string, and both the f-string and
# `cap_message`'s encode would copy it whole β under a tight
# RLIMIT_AS either allocation can raise MemoryError, and this is the
# LAST fallback, so a throw here costs the `done` frame outright and
# the run misreports as an exit or a timeout. The slice is a
# code-unit prefix, which bounds the bytes at 4x, and the following
# `cap_message` still applies the exact byte cap.
name = raw_name[:_MAX_FALLBACK_NAME_CHARS] if type(raw_name) is str else "<unknown>"
except _BaseException: # noqa: BLE001 -- a raising __name__ must not cost the done frame
name = "<unknown>"
# Wrapped for the same reason: `cap_message` encodes, and its allocation
# is the only step left that can still fail. The fixed literal needs no
# budget, so it can always be delivered.
try:
return cap_message(f"{name}: {unrenderable}", max_bytes)
except _BaseException: # noqa: BLE001 -- the done frame outranks the diagnostic's detail
return unrenderable
return safe_model_traceback
_SAFE_MODEL_TRACEBACK = _make_failure_reporter()
def _join_bounded(lines, max_bytes: int) -> str:
"""Join formatter output, stopping once the budget is comfortably passed.
``format()`` yields lines lazily; consuming it whole for an exception
carrying a huge message would materialize the full text only for
``_cap_message`` to throw it away β enough over-shoot to exhaust
``RLIMIT_AS``. Stop after the accumulated CHARACTER count passes the byte
budget (chars lower-bound UTF-8 bytes); the caller's ``_cap_message``
does the exact byte-level cut.
"""
chunks: list[str] = []
total = 0
for line in lines:
# A single yielded line can itself dwarf the budget (the exception
# message rides in one line): keep only the prefix it can ever need.
if len(line) > max_bytes + 1:
line = line[: max_bytes + 1]
chunks.append(line)
total += len(line)
if total > max_bytes:
break
return "".join(chunks)
def _done_with_value(
value: Any,
max_value_bytes: int,
# Bound as DEFAULT ARGUMENTS so they are captured at import time, before
# model code runs: `_done_with_value` runs AFTER the program (which is
# `__main__`) may have rebound `__main__._check_done_value` or
# `__main__._encode_json_plain`, and a module-global lookup at call time
# would let a one-line rebind rewrite a legitimate success into an
# `exception`. Defaults are evaluated at def time, so they are the originals.
_check_done_value: Any = _check_done_value,
_encode_json_plain: Any = _encode_json_plain,
_cap_message: Any = _cap_message,
) -> dict[str, Any] | str:
"""Build the terminal done frame under the seam's lossless-JSON contract.
A completion value returned by the program (``None`` when it returns
nothing) that is not lossless JSON fails the run as ``invalid-output``; a
serialized value beyond ``max_value_bytes`` fails as ``output-limit``.
Substituting a ``repr`` or truncated string would be a silent lie about
what the program computed, so both paths refuse instead (mirroring the
worker backend's contract). ``None`` crosses as an exact JSON ``null``.
The SUCCESS path returns the whole ``"done"`` frame as an ALREADY-ENCODED
JSON string: the admitted value is serialized here, at its validation
point, rather than handed to ``send_sync`` to re-walk later. The program
can keep mutating the returned list/dict from a daemon thread or signal
handler after it returns, so a second traversal held at a later point
would be a TOCTOU β a mutation into a non-JSON type would let that later
encode throw outside the settlement handler and downgrade the run
host-side to ``worker-exit``. Serializing once, inside the try that wraps
this call, closes the window: if a concurrent mutation makes the encode
throw, the exception handler classifies it as ``exception``, and once the
string is produced the frame is sent verbatim with no further touching of
the live value. Returns a ``dict`` only for a rejection (an error frame
carries no live model value and is safe to send via ``send_sync``).
"""
# One bounded walk folds the losslessness check and the byte meter (mirrors
# the host's checkDoneValue): the former split ran the full losslessness
# walk first, materializing one tuple per element for a wide completion
# before the size cap could reject it β an RLIMIT_AS death on a value the
# meter would have refused. The value's escaped JSON is then produced in the
# SAME call, so the admitted value is serialized exactly once (see above);
# its size the walk proved within budget. Iterative like the encoder, so a
# valid completion deeper than the recursion limit still checks.
rejection = _check_done_value(value, max_value_bytes)
if rejection is not None:
kind, message = rejection
# The rejection diagnostic is capped like an exception message: a
# reason embedding a hostile class name (a huge `type(value).__name__`)
# could otherwise make the done frame exceed the host's frame parse cap
# and be silently dropped β an invalid-output run misreported as a
# worker-exit.
return {"type": "done", "error": {"kind": kind, "message": _cap_message(message, max_value_bytes)}}
# Pre-encode the value at the validation point (not in `_run`'s later send,
# which is outside the try): see the TOCTOU note in the docstring. The value
# is JSON-plain by construction, so `_encode_json_plain` is the encoder.
return '{"type": "done", "value": ' + _encode_json_plain(value) + "}"
def main() -> None:
channel = ProtocolChannel(PROTOCOL_FD)
asyncio.run(_run(channel))
if __name__ == "__main__":
main()
|