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

1""" 

2_files_output.py — agent files-output (output.type: files) validation + write helper. 

3 

4V1-sopimus (Phase 6 doc, ### Tiedostotuloste): yksi jaettu helper palvelee 

5agent-tason files-output-käsittelijää (_llm.py). Oletuskirjoituskohde on 

6workspace-juuri (yhteinen workspace-volume /project, ADR-0024); 

7manifestin output.path rajaa kirjoituksen workspace-juuren alihakemistoon. 

8Validointi on kova (ADR-0012) ja atominen: jos yksikin entry on 

9virheellinen, mitään ei kirjoiteta ja kutsuja epäonnistuu fail-fast — 

10ei osittaisia kirjoituksia. 

11""" 

12from __future__ import annotations 

13 

14import os 

15from pathlib import Path 

16from typing import Any 

17 

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

19# 10 MiB yhteensä per step. 

20DEFAULT_MAX_FILE_BYTES = 1_048_576 

21DEFAULT_MAX_TOTAL_BYTES = 10_485_760 

22 

23 

24class FilesOutputError(ValueError): 

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

26 

27 

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

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

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

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

32 

33 

34def validate_and_write_files( 

35 entries: Any, 

36 workspace_root: Path, 

37 target_path: str | None = None, 

38) -> None: 

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

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

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

42 if not isinstance(entries, list): 

43 raise FilesOutputError( 

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

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

46 ) 

47 root = workspace_root.resolve(strict=False) 

48 base = _resolve_target_path(root, target_path) 

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

50 total_bytes = 0 

51 for index, entry in enumerate(entries): 

52 ref = f"entry {index}" 

53 path, content = _validate_entry(entry, ref, base, root) 

54 try: 

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

56 except UnicodeEncodeError as exc: 

57 raise FilesOutputError( 

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

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

60 ) from exc 

61 if len(encoded) > DEFAULT_MAX_FILE_BYTES: 

62 raise FilesOutputError( 

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

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

65 ) 

66 total_bytes += len(encoded) 

67 if total_bytes > DEFAULT_MAX_TOTAL_BYTES: 

68 raise FilesOutputError( 

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

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

71 ) 

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

73 for path, encoded, ref in prepared: 

74 try: 

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

76 path.write_bytes(encoded) 

77 except OSError as exc: 

78 raise FilesOutputError( 

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

80 ) from exc 

81 

82 

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

84 if not isinstance(entry, dict): 

85 raise FilesOutputError( 

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

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

88 ) 

89 raw_path = entry.get("path") 

90 content = entry.get("content") 

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

92 raise FilesOutputError( 

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

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

95 ) 

96 if not raw_path.strip(): 

97 raise FilesOutputError( 

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

99 f"is required (ADR-0012)" 

100 ) 

101 return _resolve_path(root, raw_path, base, ref), content 

102 

103 

104def _resolve_target_path(root: Path, target_path: str | None) -> Path: 

105 if target_path is None: 

106 return root 

107 if not target_path.strip(): 

108 raise FilesOutputError( 

109 "files output: output.path is empty or whitespace-only (ADR-0012)" 

110 ) 

111 return _resolve_path(root, target_path, root, "output.path") 

112 

113 

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

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

116 

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

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

119 """ 

120 candidate = Path(raw_path) 

121 if ".." in candidate.parts: 

122 raise FilesOutputError( 

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

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

125 ) 

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

127 resolved = target.resolve(strict=False) 

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

129 raise FilesOutputError( 

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

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

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

133 ) 

134 relative_parts = resolved.relative_to(root).parts 

135 if ".git" in relative_parts: 

136 raise FilesOutputError( 

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

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

139 ) 

140 return resolved