Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions CONTRIBUTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,18 @@ make integration # real-PostgreSQL integration/security harness
make security-gate # M0 security veto gate
```

Run `make typecheck` and `make test` sequentially, never concurrently.
`make test` depends on `bot-build`, and
`tests/unit/test_bot_delivery_model_egress_contract.py` also invokes
`npm run build`. Both can write TypeScript build trees while `make typecheck`
reads them through `tsc`; overlap can therefore produce a spurious test
failure.

Separately, a single non-reproducible `test_egress_grant` failure remains an
undiagnosed watch item and is not attributed to the build-tree collision above.
Do not treat it as a known defect without a fresh reproduction; split a
recurrence into its own issue with that evidence.

When you are done, stop the harness:

```bash
Expand Down
52 changes: 42 additions & 10 deletions applications/compiler_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
import subprocess
import sys
from pathlib import Path, PurePath
from typing import Final, Protocol, cast
from typing import Final, Never, Protocol, cast

from adapters.parsers.ragflow_markdown import compile_rich_markdown, rich_token_count
from engine.supply import (
Expand Down Expand Up @@ -45,6 +45,11 @@ def __call__(
) -> CompilationOutcome: ...


class _PrivacySafeArgumentParser(argparse.ArgumentParser):
def error(self, message: str) -> Never:
raise SystemExit("compiler runner arguments are invalid")


def _boundary_failure() -> CompilationFailure:
return CompilationFailure(
code=CompilationFailureCode.UNSUPPORTED_DOCUMENT_SHAPE,
Expand Down Expand Up @@ -202,7 +207,7 @@ def _emit(


def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser = _PrivacySafeArgumentParser(description=__doc__)
parser.add_argument("--compile", action="store_true")
parser.add_argument("--config", default="markdown-config-v3")
parser.add_argument(
Expand Down Expand Up @@ -237,11 +242,17 @@ def _parser() -> argparse.ArgumentParser:


def _safe_markdown_files(root: Path) -> tuple[Path, ...]:
if not root.is_dir():
raise ValueError("acceptance root must be a directory")
if root.is_symlink() or not root.is_dir():
raise SystemExit("acceptance root must be a non-symlink directory")
return tuple(
sorted(
(path for path in root.rglob("*.md") if path.is_file()),
(
path
for path in root.rglob("*", recurse_symlinks=False)
if path.name.casefold().endswith(".md")
and not path.is_symlink()
and path.is_file()
),
key=lambda path: PurePath(*path.relative_to(root).parts).as_posix(),
)
)
Expand Down Expand Up @@ -330,16 +341,28 @@ def _write_acceptance_report(
*,
acceptance_context: _AcceptanceContext,
) -> None:
if ".context-engine" not in output.parts:
raise ValueError("acceptance reports must be written under .context-engine")
try:
state_index = len(output.parts) - 1 - output.parts[::-1].index(
".context-engine"
)
except ValueError:
raise SystemExit(
"acceptance reports must be written under .context-engine"
) from None
state_directory = Path(*output.parts[: state_index + 1]).resolve()
resolved_output = output.resolve()
if resolved_output == state_directory or not resolved_output.is_relative_to(
state_directory
):
raise SystemExit("acceptance reports must be written under .context-engine")
report = _acceptance_report(
root,
token_ceiling,
acceptance_context=acceptance_context,
)
serialized = json.dumps(report, sort_keys=True, separators=(",", ":")) + "\n"
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(serialized, encoding="utf-8")
resolved_output.parent.mkdir(parents=True, exist_ok=True)
resolved_output.write_text(serialized, encoding="utf-8")
sys.stdout.write(serialized)


Expand All @@ -358,6 +381,8 @@ def main() -> None:
if args.acceptance_report:
if args.root is None:
raise SystemExit("--acceptance-report requires --root")
if cast(int, args.token_ceiling) < 1:
raise SystemExit("rich Markdown token ceiling must be positive")
_write_acceptance_report(
cast(Path, args.root),
cast(Path, args.output),
Expand All @@ -368,5 +393,12 @@ def main() -> None:
raise SystemExit("one runner operation is required")


def _privacy_safe_main() -> None:
try:
main()
except Exception:
raise SystemExit("compiler runner operation failed") from None
Comment on lines +399 to +400

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Catch interrupts at the privacy boundary

stometa, when an operator interrupts a long corpus scan with Ctrl-C, Python raises KeyboardInterrupt, which inherits directly from BaseException and therefore bypasses this handler. The interpreter then emits a traceback containing machine-local absolute paths such as applications/compiler_runner.py, defeating the process-level output-privacy guarantee; handle KeyboardInterrupt before it escapes while continuing to preserve the deliberate SystemExit messages.

Useful? React with 👍 / 👎.



if __name__ == "__main__":
main()
_privacy_safe_main()
67 changes: 67 additions & 0 deletions tests/unit/test_compiler_runner_acceptance_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import pytest

import applications.compiler_runner as compiler_runner
from eval._compiler_acceptance import acceptance_context
from eval.embedding_benchmark import (
BenchmarkUnavailable,
validate_json_schema_document,
Expand Down Expand Up @@ -231,6 +232,72 @@ def test_acceptance_report_is_count_only_deterministic_and_written_under_ignore(
assert first.stdout == second.stdout


def test_acceptance_corpus_matches_file_provider_markdown_directory_rules(
tmp_path: Path,
) -> None:
corpus = tmp_path / "corpus"
corpus.mkdir()
lowercase_directory = corpus / "bare-lowercase"
lowercase_directory.mkdir()
bare_lowercase = lowercase_directory / ".md"
bare_lowercase.write_text("# Bare lowercase\n", encoding="utf-8")
uppercase_directory = corpus / "bare-uppercase"
uppercase_directory.mkdir()
bare_uppercase = uppercase_directory / ".MD"
bare_uppercase.write_text("# Bare uppercase\n", encoding="utf-8")
included = corpus / "included.MD"
included.write_text("# Included\n", encoding="utf-8")
ordinary = corpus / "ordinary.md"
ordinary.write_text("# Ordinary\n", encoding="utf-8")
target = corpus / "target.md"
target.write_text("# Target\n", encoding="utf-8")
(corpus / "linked.md").symlink_to(target)
external = tmp_path / "external"
external.mkdir()
(external / "outside.md").write_text("# Outside\n", encoding="utf-8")
(corpus / "linked-directory").symlink_to(external, target_is_directory=True)

discovered = compiler_runner._safe_markdown_files(corpus)

assert discovered == (
bare_lowercase,
bare_uppercase,
included,
ordinary,
target,
)


def test_acceptance_corpus_root_must_not_be_a_symlink(tmp_path: Path) -> None:
corpus = tmp_path / "corpus"
corpus.mkdir()
linked_root = tmp_path / "linked-root"
linked_root.symlink_to(corpus, target_is_directory=True)

with pytest.raises(SystemExit, match="non-symlink directory"):
compiler_runner._safe_markdown_files(linked_root)


def test_acceptance_output_must_resolve_beneath_its_state_directory(
tmp_path: Path,
) -> None:
corpus = tmp_path / "corpus"
corpus.mkdir()
state_directory = tmp_path / ".context-engine"
state_directory.mkdir()
escaped_output = state_directory / ".." / "report.json"

with pytest.raises(SystemExit, match="under .context-engine"):
compiler_runner._write_acceptance_report(
corpus,
escaped_output,
2048,
acceptance_context=acceptance_context(),
)

assert not (tmp_path / "report.json").exists()


@pytest.mark.parametrize(
"failure_kind",
("io", "permission", "vanished", "directory"),
Expand Down
165 changes: 165 additions & 0 deletions tests/unit/test_compiler_runner_cli_privacy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
from __future__ import annotations

import re
import subprocess
import sys
from pathlib import Path

import pytest

REPOSITORY_ROOT = Path(__file__).parents[2]
_MACHINE_ABSOLUTE_PATH_PATTERNS = (
re.compile(r"(?<![:A-Za-z0-9_])/(?:[^\s/]+/)*[^\s/]+"),
re.compile(r"(?<![A-Za-z0-9_])[A-Za-z]:/(?:[^\s/]+/)*[^\s/]+"),
re.compile(r"(?<![A-Za-z0-9_])[A-Za-z]:\\(?:[^\s\\]+\\)*[^\s\\]+"),
re.compile(r"(?<!\\)\\\\[^\\\r\n]+\\[^\\\r\n]+"),
)


def _run_acceptance_cli(*arguments: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[sys.executable, "-m", "applications.compiler_runner", *arguments],
cwd=REPOSITORY_ROOT,
check=False,
capture_output=True,
text=True,
timeout=30,
)


def _assert_process_output_is_private(
completed: subprocess.CompletedProcess[str],
) -> None:
captured = completed.stdout + completed.stderr

assert str(REPOSITORY_ROOT) not in captured
assert all(
pattern.search(captured) is None
for pattern in _MACHINE_ABSOLUTE_PATH_PATTERNS
)


def test_machine_absolute_path_guard_recognizes_platform_path_shapes() -> None:
canaries = (
"/private-machine/repository/app.py",
"C:/" + "Users" + "/person/repository/app.py",
"C:\\Users\\person\\repository\\app.py",
"\\\\server\\share\\repository\\app.py",
)

assert all(
any(
pattern.search(canary) is not None
for pattern in _MACHINE_ABSOLUTE_PATH_PATTERNS
)
for canary in canaries
)


def test_process_output_privacy_assertion_rejects_a_planted_leak() -> None:
completed = subprocess.CompletedProcess(
args=(),
returncode=1,
stdout="C:/" + "Users" + "/person/repository/app.py\n",
stderr="",
)

with pytest.raises(AssertionError):
_assert_process_output_is_private(completed)


def test_acceptance_cli_success_output_has_no_machine_local_absolute_path(
tmp_path: Path,
) -> None:
corpus = tmp_path / "corpus"
corpus.mkdir()
(corpus / "note.md").write_text("# Note\n\nBody.\n", encoding="utf-8")
output = tmp_path / ".context-engine/report.json"

completed = _run_acceptance_cli(
"--acceptance-report",
"--root",
str(corpus),
"--output",
str(output),
)

assert completed.returncode == 0
_assert_process_output_is_private(completed)


def test_acceptance_cli_parser_error_does_not_echo_the_invalid_argument() -> None:
invalid_argument = "/private-machine/repository/not-an-integer"

completed = _run_acceptance_cli("--token-ceiling", invalid_argument)

assert completed.returncode != 0
assert invalid_argument not in completed.stdout + completed.stderr
_assert_process_output_is_private(completed)


def test_five_acceptance_cli_operator_errors_emit_their_specific_safe_messages(
tmp_path: Path,
) -> None:
corpus = tmp_path / "corpus"
corpus.mkdir()
(corpus / "note.md").write_text("# Note\n\nBody.\n", encoding="utf-8")
root_file = tmp_path / "root-file.md"
root_file.write_text("# Not a directory\n", encoding="utf-8")
state_directory = tmp_path / ".context-engine"
state_directory.mkdir()
operator_errors = (
(
("--root", str(tmp_path / "missing")),
"acceptance root must be a non-symlink directory",
),
(
("--root", str(root_file)),
"acceptance root must be a non-symlink directory",
),
(
("--root", str(corpus), "--output", str(tmp_path / "report.json")),
"acceptance reports must be written under .context-engine",
),
(
("--root", str(corpus), "--token-ceiling", "0"),
"rich Markdown token ceiling must be positive",
),
(
("--root", str(corpus), "--output", str(state_directory)),
"acceptance reports must be written under .context-engine",
),
)

for arguments, expected_message in operator_errors:
completed = _run_acceptance_cli("--acceptance-report", *arguments)

assert completed.returncode != 0
assert completed.stdout == ""
assert completed.stderr == f"{expected_message}\n"
_assert_process_output_is_private(completed)


def test_acceptance_cli_unexpected_failure_emits_only_the_generic_safe_message(
tmp_path: Path,
) -> None:
corpus = tmp_path / "corpus"
corpus.mkdir()
(corpus / "note.md").write_text("# Note\n\nBody.\n", encoding="utf-8")
state_directory = tmp_path / ".context-engine"
state_directory.mkdir()
blocked_parent = state_directory / "blocked-parent"
blocked_parent.write_text("not a directory\n", encoding="utf-8")

completed = _run_acceptance_cli(
"--acceptance-report",
"--root",
str(corpus),
"--output",
str(blocked_parent / "report.json"),
)

assert completed.returncode != 0
assert completed.stdout == ""
assert completed.stderr == "compiler runner operation failed\n"
_assert_process_output_is_private(completed)
Loading