|
| 1 | +"""Bounded, pinned implementation evidence for subscription module cards.""" |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +import hashlib |
| 5 | +import json |
| 6 | +from types import SimpleNamespace |
| 7 | + |
| 8 | +import pytest |
| 9 | + |
| 10 | +from infermatrix_copilot.kb_service.init_modules import ( |
| 11 | + MAX_CARD_BYTES, SYSTEM_CARD, SYSTEM_CARD_IMPL, _implementation_payload, _Modules, |
| 12 | +) |
| 13 | +from infermatrix_copilot.kb_service.init_stages import _fence, run_stage |
| 14 | +from infermatrix_copilot.kb_service.models import ModelRole |
| 15 | +from test_kb_init_modules import CardGateway, _modules_lifecycle, _skeleton, _world_with_tools |
| 16 | +from test_kb_init_skeleton import _commit, _runtime, world # noqa: F401 |
| 17 | + |
| 18 | +PIN = "a" * 40 |
| 19 | + |
| 20 | + |
| 21 | +def _index(files): |
| 22 | + return SimpleNamespace(identity={"pin": PIN}, entries={ |
| 23 | + path: {"status": "ready", "lines": text.splitlines(), "text": text, |
| 24 | + "sha256": hashlib.sha256(text.encode()).hexdigest()} for path, text in files.items()}) |
| 25 | + |
| 26 | + |
| 27 | +def _payload(paths): |
| 28 | + return {"files": [{"path": path} for path in paths], "doc_files": [], "groups": {}} |
| 29 | + |
| 30 | + |
| 31 | +def test_complete_evidence_includes_late_runtime_and_conditional_defaults(): |
| 32 | + files = {"scripts/bridge.js": "#!/usr/bin/env node\n" + "// comment\n" * 190 |
| 33 | + + "new WebSocketServer({port:19600});\n", |
| 34 | + "scripts/teardown.sh": "STOP_SERVICE=0\nDISABLE_SERVICE=0\n" |
| 35 | + "if [ \"$STOP_SERVICE\" = 1 ]; then systemctl stop nfs; fi\n" |
| 36 | + "echo 'Manually run hvigorw build'\n"} |
| 37 | + output = _implementation_payload(_payload(files), _index(files), PIN) |
| 38 | + assert output["source_pin"] == PIN |
| 39 | + for item in output["source_evidence"]: |
| 40 | + assert item["status"] == "complete" |
| 41 | + assert item["sha256"] == hashlib.sha256(files[item["path"]].encode()).hexdigest() |
| 42 | + shown = item["ranges"][0] |
| 43 | + assert (shown["start"], shown["end"]) == (1, len(files[item["path"]].splitlines())) |
| 44 | + assert shown["text"] == "\n".join(f"{n}: {line}" for n, line in |
| 45 | + enumerate(files[item["path"]].splitlines(), 1)) |
| 46 | + assert "new WebSocketServer" in output["source_evidence"][0]["ranges"][0]["text"] |
| 47 | + |
| 48 | + |
| 49 | +def test_capped_multibyte_evidence_is_fair_numbered_and_explicitly_partial(): |
| 50 | + files = {"scripts/large.js": "\n".join(f"const 中文{n} = '<value>';" for n in range(5000)), |
| 51 | + "scripts/small.js": "runtime_optional_bridge();\n"} |
| 52 | + output = _implementation_payload(_payload(files), _index(files), PIN) |
| 53 | + assert len(_fence(output).encode()) <= MAX_CARD_BYTES |
| 54 | + large, small = output["source_evidence"] |
| 55 | + assert large["status"] == "partial" and large["total_lines"] == 5000 |
| 56 | + assert large["ranges"][0]["start"] == 1 |
| 57 | + assert large["ranges"][-1]["end"] == 5000 |
| 58 | + assert large["ranges"][0]["end"] < large["ranges"][-1]["start"] |
| 59 | + assert small["status"] == "complete" # large first file cannot starve a later small file |
| 60 | + for shown in large["ranges"]: |
| 61 | + lines = files[large["path"]].splitlines() |
| 62 | + assert shown["text"] == "\n".join(f"{n}: {lines[n - 1]}" for n in |
| 63 | + range(shown["start"], shown["end"] + 1)) |
| 64 | + |
| 65 | + |
| 66 | +def test_failed_or_missing_indexed_source_stays_unknown_without_replacement_text(): |
| 67 | + index = _index({}) |
| 68 | + index.entries["bad.js"] = {"status": "read_failed", "text": "invalid replacement text"} |
| 69 | + output = _implementation_payload(_payload(["bad.js", "missing.js"]), index, PIN) |
| 70 | + assert all(item["status"] == "unknown" and not item["ranges"] for item in output["source_evidence"]) |
| 71 | + assert "invalid replacement text" not in json.dumps(output) |
| 72 | + |
| 73 | + |
| 74 | +def test_metadata_cap_and_pinned_index_identity_fail_closed(): |
| 75 | + with pytest.raises(ValueError, match="byte cap"): |
| 76 | + _implementation_payload({**_payload([]), "doc_files": ["x" * MAX_CARD_BYTES]}, _index({}), PIN) |
| 77 | + with pytest.raises(ValueError, match="fixed source SHA"): |
| 78 | + _implementation_payload(_payload([]), _index({}), "b" * 40) |
| 79 | + |
| 80 | + |
| 81 | +def test_unlimited_prompt_identity_and_instructions_leave_legacy_unchanged(): |
| 82 | + assert _Modules._input_options(SimpleNamespace(rt=SimpleNamespace(unlimited_subscription=False))) == {} |
| 83 | + assert _Modules._input_options(SimpleNamespace(rt=SimpleNamespace(unlimited_subscription=True))) == { |
| 84 | + "module_prompt_version": 2} |
| 85 | + assert "never the full code" in SYSTEM_CARD and "source_pin" not in SYSTEM_CARD |
| 86 | + assert "never the full code" not in SYSTEM_CARD_IMPL |
| 87 | + assert all(text in SYSTEM_CARD_IMPL for text in |
| 88 | + ("default side effects", "printed/manual", "Unshown behavior remains unknown", "steps as executed")) |
| 89 | + |
| 90 | + |
| 91 | +class ImplementationGateway(CardGateway): |
| 92 | + def subscription_billing(self, role): |
| 93 | + return True |
| 94 | + |
| 95 | + def call_json(self, role, *, system, prompt, **kwargs): |
| 96 | + reply = super().call_json(role, system=SYSTEM_CARD if system == SYSTEM_CARD_IMPL else system, |
| 97 | + prompt=prompt, **kwargs) |
| 98 | + if system == SYSTEM_CARD_IMPL: |
| 99 | + self.calls[-1]["system"] = system |
| 100 | + return reply |
| 101 | + |
| 102 | + |
| 103 | +def test_official_unlimited_modules_use_actual_pinned_source_bodies(world): |
| 104 | + _world_with_tools(world) |
| 105 | + source = "def secret():\n return optional_runtime_service()\n" + "\n".join( |
| 106 | + f"def another_{n}(): pass" for n in range(8)) + "\n" |
| 107 | + _commit(world["upstream"], {"tools/lint/runtime.py": source}, "runtime helper") |
| 108 | + _skeleton(world) |
| 109 | + gateway = ImplementationGateway() |
| 110 | + record = run_stage(_runtime(world, gateway, generator=ModelRole("generator", "zcode", "GLM-5.3")), |
| 111 | + _modules_lifecycle(), "modules", dry_run=True, unlimited_subscription=True) |
| 112 | + assert record.status == "dry_run", record.problems |
| 113 | + calls = [call for call in gateway.calls if call["system"] == SYSTEM_CARD_IMPL] |
| 114 | + assert calls and all(len(call["prompt"].encode()) <= MAX_CARD_BYTES for call in calls) |
| 115 | + payloads = [json.loads(call["prompt"].split("\n", 1)[1].rsplit("</untrusted_data>", 1)[0]) for call in calls] |
| 116 | + item = next(item for payload in payloads for item in payload["source_evidence"] |
| 117 | + if item["path"] == "tools/lint/runtime.py") |
| 118 | + assert item["status"] == "complete" and item["sha256"] == hashlib.sha256(source.encode()).hexdigest() |
| 119 | + assert "2: return optional_runtime_service()" in item["ranges"][0]["text"] |
| 120 | + assert all(payload["source_pin"] == record.pin for payload in payloads) |
| 121 | + assert all(len(file.get("signatures", [])) <= 4 for payload in payloads for file in payload["files"]) |
| 122 | + |
| 123 | + |
| 124 | +def test_unlimited_card_cannot_name_a_module_file_outside_its_offered_subset(world, monkeypatch): |
| 125 | + import infermatrix_copilot.kb_service.init_modules as modules |
| 126 | + monkeypatch.setattr(modules, "MAX_CARD_FILES", 1) |
| 127 | + _world_with_tools(world) |
| 128 | + _commit(world["upstream"], {"tools/lint/a.py": "def first(): pass\n"}, "another lint file") |
| 129 | + _skeleton(world) |
| 130 | + |
| 131 | + class UnofferedGateway(ImplementationGateway): |
| 132 | + def call_json(self, role, *, system, prompt, **kwargs): |
| 133 | + reply = super().call_json(role, system=system, prompt=prompt, **kwargs) |
| 134 | + if system == SYSTEM_CARD_IMPL and '"module": "tools/lint/"' in prompt: |
| 135 | + reply.data["key_files"] = [{"path": "tools/lint/y.py", "what": "unsupported unseen file"}] |
| 136 | + return reply |
| 137 | + |
| 138 | + record = run_stage(_runtime(world, UnofferedGateway(), |
| 139 | + generator=ModelRole("generator", "zcode", "GLM-5.3")), |
| 140 | + _modules_lifecycle(), "modules", dry_run=True, unlimited_subscription=True) |
| 141 | + assert record.status == "dry_run", record.problems |
| 142 | + assert any("key_files 'tools/lint/y.py' is not offered" in note for note in record.notes) |
0 commit comments