Coverage for core/langgraph/_files_output.py: 95%

1""" 

2_files_output.py — jaettu files-validation/kirjoitus-helper. 

3 

4Yksi jaettu helper palvelee live `files-write`-builtinia; `FilesOutputError` 

5on jaettu `files-delete`-builtinin kanssa. Kirjoituskohde on workspace-juuri 

6(yhteinen workspace-volume /project, ADR-0024). Validointi on kova (ADR-0012) 

7ja atominen: jos yksikin entry on virheellinen, mitään ei kirjoiteta ja 

8kutsuja epäonnistuu fail-fast — ei osittaisia kirjoituksia. 

9""" 

10from __future__ import annotations 

11 

12import os 

13from pathlib import Path 

14from typing import Any 

15 

16# Deterministiset oletuskatot (Phase 7 -valinta): 1 MiB per entry, 

17# 10 MiB yhteensä per step. 

18DEFAULT_MAX_FILE_BYTES = 1_048_576 

19DEFAULT_MAX_TOTAL_BYTES = 10_485_760 

20 

21 

22class FilesOutputError(ValueError): 

23 """Kova files-output-virhe (ADR-0012) — nimeää virheellisen entryn.""" 

24 

25 

26def resolve_workspace_root(state: dict) -> Path: 

27 """Workspace-juuri: state.project_root (builtin-konventio) → 

28 PROJECT_ROOT-env → /project (ADR-0024).""" 

29 return Path(state.get("project_root") or os.getenv("PROJECT_ROOT", "/project")) 

30 

31 

32def validate_and_write_files( 

33 entries: Any, 

34 workspace_root: Path, 

35) -> None: 

36 """Validoi KAIKKI entryt atomisesti ja kirjoittaa sitten (create/modify, 

37 UTF-8). Yksikin virheellinen entry → FilesOutputError entryn nimellä — 

38 mitään ei kirjoiteta (fail-fast, ADR-0012).""" 

39 if not isinstance(entries, list): 

40 raise FilesOutputError( 

41 f"files output must be a JSON list of {{path, content}} entries, " 

42 f"got {type(entries).__name__} (ADR-0012)" 

43 ) 

44 root = workspace_root.resolve(strict=False) 

45 prepared: list[tuple[Path, bytes, str]] = [] 

46 total_bytes = 0 

47 for index, entry in enumerate(entries): 

48 ref = f"entry {index}" 

49 path, content = _validate_entry(entry, ref, root) 

50 try: 

51 encoded = content.encode("utf-8") 

52 except UnicodeEncodeError as exc: 

53 raise FilesOutputError( 

54 f"files output {ref}: content is not UTF-8 text " 

55 f"(path '{path}'): {exc} (ADR-0012)" 

56 ) from exc 

57 if len(encoded) > DEFAULT_MAX_FILE_BYTES: 

58 raise FilesOutputError( 

59 f"files output {ref}: path '{path}' is {len(encoded)} bytes, " 

60 f"exceeding the per-file cap of {DEFAULT_MAX_FILE_BYTES} bytes (ADR-0012)" 

61 ) 

62 total_bytes += len(encoded) 

63 if total_bytes > DEFAULT_MAX_TOTAL_BYTES: 

64 raise FilesOutputError( 

65 f"files output {ref}: path '{path}' exceeds the total cap of " 

66 f"{DEFAULT_MAX_TOTAL_BYTES} bytes per step (ADR-0012)" 

67 ) 

68 prepared.append((path, encoded, ref)) 

69 for path, encoded, ref in prepared: 

70 try: 

71 path.parent.mkdir(parents=True, exist_ok=True) 

72 path.write_bytes(encoded) 

73 except OSError as exc: 

74 raise FilesOutputError( 

75 f"files output {ref}: failed to write '{path}': {exc} (ADR-0012)" 

76 ) from exc 

77 

78 

79def _validate_entry(entry: Any, ref: str, root: Path) -> tuple[Path, str]: 

80 if not isinstance(entry, dict): 

81 raise FilesOutputError( 

82 f"files output {ref}: expected an object {{path, content}}, " 

83 f"got {type(entry).__name__} (ADR-0012)" 

84 ) 

85 raw_path = entry.get("path") 

86 content = entry.get("content") 

87 if not isinstance(raw_path, str) or not isinstance(content, str): 

88 raise FilesOutputError( 

89 f"files output {ref}: each entry must have a string 'path' and a " 

90 f"string 'content' (ADR-0012)" 

91 ) 

92 if not raw_path.strip(): 

93 raise FilesOutputError( 

94 f"files output {ref}: path is empty or whitespace-only — a path " 

95 f"is required (ADR-0012)" 

96 ) 

97 return _resolve_path(root, raw_path, root, ref), content 

98 

99 

100def _resolve_path(root: Path, raw_path: str, base: Path, ref: str) -> Path: 

101 """Resolvoi polku workspace-juuren sisään ja palauttaa kohdepolun. 

102 

103 Kova validointi (ADR-0012): ei `..`-poikkeamaa, ei absoluuttista polkua 

104 juuren ulkopuolelle, ei symlink-pakoa (resolve), ei .git-kohdetta. 

105 """ 

106 candidate = Path(raw_path) 

107 if ".." in candidate.parts: 

108 raise FilesOutputError( 

109 f"files output {ref}: path '{raw_path}' contains '..' — path " 

110 f"traversal is forbidden (ADR-0012)" 

111 ) 

112 target = candidate if candidate.is_absolute() else base / candidate 

113 resolved = target.resolve(strict=False) 

114 if resolved != root and root not in resolved.parents: 

115 raise FilesOutputError( 

116 f"files output {ref}: path '{raw_path}' resolves outside the " 

117 f"workspace root '{root}' — absolute paths outside the root and " 

118 f"symlink escapes are forbidden (ADR-0012)" 

119 ) 

120 relative_parts = resolved.relative_to(root).parts 

121 if ".git" in relative_parts: 

122 raise FilesOutputError( 

123 f"files output {ref}: path '{raw_path}' targets the .git " 

124 f"directory — forbidden as a write target (ADR-0012)" 

125 ) 

126 return resolved