+718

apps/import/facet_ingest.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + import hashlib 7 7 + import json 8 8 + import re 9 9 + from datetime import datetime, timezone 10 10 + from pathlib import Path, PurePosixPath 11 11 + from typing import Any 12 12 + 13 13 + from think.entities.observations import load_observations, save_observations 14 14 + from think.entities.relationships import ( 15 15 + load_facet_relationship, 16 16 + save_facet_relationship, 17 17 + ) 18 18 + 19 19 + from .ingest import _append_decision 20 20 + 21 21 + _DAY_JSONL_RE = re.compile(r"^\d{8}\.jsonl$") 22 22 + _DAY_MD_RE = re.compile(r"^\d{8}\.md$") 23 23 + _FACET_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*$") 24 24 + _ENTITY_FILE_TYPES = { 25 25 + "entity_relationship", 26 26 + "entity_observations", 27 27 + "detected_entities", 28 28 + "activity_records", 29 29 + } 30 30 + 31 31 + 32 32 + def _read_jsonl(path: Path) -> list[dict[str, Any]]: 33 33 + if not path.is_file(): 34 34 + return [] 35 35 + 36 36 + items: list[dict[str, Any]] = [] 37 37 + with open(path, encoding="utf-8") as handle: 38 38 + for line in handle: 39 39 + line = line.strip() 40 40 + if not line: 41 41 + continue 42 42 + items.append(json.loads(line)) 43 43 + return items 44 44 + 45 45 + 46 46 + def _append_jsonl(path: Path, items: list[dict[str, Any]]) -> None: 47 47 + if not items: 48 48 + return 49 49 + 50 50 + path.parent.mkdir(parents=True, exist_ok=True) 51 51 + with open(path, "a", encoding="utf-8") as handle: 52 52 + for item in items: 53 53 + handle.write(json.dumps(item, ensure_ascii=False) + "\n") 54 54 + 55 55 + 56 56 + def _remap_entity_id(entity_id: str, id_map: dict[str, str]) -> str | None: 57 57 + return id_map.get(entity_id) 58 58 + 59 59 + 60 60 + def _parse_path(path_str: str, file_type: str) -> tuple[PurePosixPath, dict[str, str]]: 61 61 + path = PurePosixPath(path_str) 62 62 + if not path_str or path.is_absolute(): 63 63 + raise ValueError("Invalid path") 64 64 + 65 65 + if any(part in {"", ".", ".."} for part in path.parts): 66 66 + raise ValueError("Invalid path") 67 67 + 68 68 + parts = path.parts 69 69 + 70 70 + if file_type == "facet_json": 71 71 + if parts != ("facet.json",): 72 72 + raise ValueError("facet_json path must be facet.json") 73 73 + return path, {} 74 74 + 75 75 + if file_type == "entity_relationship": 76 76 + if len(parts) != 3 or parts[0] != "entities" or parts[2] != "entity.json": 77 77 + raise ValueError("entity_relationship path must be entities/<id>/entity.json") 78 78 + return path, {"entity_id": parts[1]} 79 79 + 80 80 + if file_type == "entity_observations": 81 81 + if ( 82 82 + len(parts) != 3 83 83 + or parts[0] != "entities" 84 84 + or parts[2] != "observations.jsonl" 85 85 + ): 86 86 + raise ValueError( 87 87 + "entity_observations path must be entities/<id>/observations.jsonl" 88 88 + ) 89 89 + return path, {"entity_id": parts[1]} 90 90 + 91 91 + if file_type == "detected_entities": 92 92 + if len(parts) != 2 or parts[0] != "entities" or not _DAY_JSONL_RE.match(parts[1]): 93 93 + raise ValueError("detected_entities path must be entities/YYYYMMDD.jsonl") 94 94 + return path, {"day_file": parts[1]} 95 95 + 96 96 + if file_type == "activity_config": 97 97 + if parts != ("activities", "activities.jsonl"): 98 98 + raise ValueError("activity_config path must be activities/activities.jsonl") 99 99 + return path, {} 100 100 + 101 101 + if file_type == "activity_records": 102 102 + if len(parts) != 2 or parts[0] != "activities" or not _DAY_JSONL_RE.match(parts[1]): 103 103 + raise ValueError("activity_records path must be activities/YYYYMMDD.jsonl") 104 104 + return path, {"day_file": parts[1]} 105 105 + 106 106 + if file_type == "activity_output": 107 107 + if len(parts) < 4 or parts[0] != "activities" or not re.match(r"^\d{8}$", parts[1]): 108 108 + raise ValueError( 109 109 + "activity_output path must be activities/YYYYMMDD/<activity_id>/..." 110 110 + ) 111 111 + return path, {"output_dir": str(PurePosixPath(*parts[:3]))} 112 112 + 113 113 + if file_type == "todos": 114 114 + if len(parts) != 2 or parts[0] != "todos" or not _DAY_JSONL_RE.match(parts[1]): 115 115 + raise ValueError("todos path must be todos/YYYYMMDD.jsonl") 116 116 + return path, {"day_file": parts[1]} 117 117 + 118 118 + if file_type == "calendar": 119 119 + if len(parts) != 2 or parts[0] != "calendar" or not _DAY_JSONL_RE.match(parts[1]): 120 120 + raise ValueError("calendar path must be calendar/YYYYMMDD.jsonl") 121 121 + return path, {"day_file": parts[1]} 122 122 + 123 123 + if file_type == "news": 124 124 + if len(parts) != 2 or parts[0] != "news" or not _DAY_MD_RE.match(parts[1]): 125 125 + raise ValueError("news path must be news/YYYYMMDD.md") 126 126 + return path, {"news_file": parts[1]} 127 127 + 128 128 + if file_type == "logs": 129 129 + if len(parts) != 2 or parts[0] != "logs" or not _DAY_JSONL_RE.match(parts[1]): 130 130 + raise ValueError("logs path must be logs/YYYYMMDD.jsonl") 131 131 + return path, {"day_file": parts[1]} 132 132 + 133 133 + raise ValueError(f"Unsupported file type: {file_type}") 134 134 + 135 135 + 136 136 + def _decode_text(raw_bytes: bytes) -> str: 137 137 + return raw_bytes.decode("utf-8") 138 138 + 139 139 + 140 140 + def _parse_json_bytes(raw_bytes: bytes) -> Any: 141 141 + return json.loads(_decode_text(raw_bytes)) 142 142 + 143 143 + 144 144 + def _parse_jsonl_bytes(raw_bytes: bytes) -> list[dict[str, Any]]: 145 145 + items: list[dict[str, Any]] = [] 146 146 + for line_number, line in enumerate(_decode_text(raw_bytes).splitlines(), start=1): 147 147 + line = line.strip() 148 148 + if not line: 149 149 + continue 150 150 + try: 151 151 + value = json.loads(line) 152 152 + except json.JSONDecodeError as exc: 153 153 + raise ValueError(f"Invalid JSONL at line {line_number}: {exc.msg}") from exc 154 154 + if not isinstance(value, dict): 155 155 + raise ValueError(f"Invalid JSONL at line {line_number}: item must be an object") 156 156 + items.append(value) 157 157 + return items 158 158 + 159 159 + 160 160 + def _check_unmapped_entities( 161 161 + data: Any, 162 162 + id_map: dict[str, str], 163 163 + file_type: str, 164 164 + path_info: dict[str, str], 165 165 + ) -> list[str]: 166 166 + unmapped: list[str] = [] 167 167 + 168 168 + def add(entity_id: str) -> None: 169 169 + if entity_id and _remap_entity_id(entity_id, id_map) is None and entity_id not in unmapped: 170 170 + unmapped.append(entity_id) 171 171 + 172 172 + if file_type in {"entity_relationship", "entity_observations"}: 173 173 + add(path_info["entity_id"]) 174 174 + elif file_type == "detected_entities": 175 175 + for item in data: 176 176 + entity_id = item.get("id") 177 177 + if entity_id: 178 178 + add(str(entity_id)) 179 179 + elif file_type == "activity_records": 180 180 + for item in data: 181 181 + active_entities = item.get("active_entities") 182 182 + if not isinstance(active_entities, list): 183 183 + continue 184 184 + for entity_id in active_entities: 185 185 + if entity_id: 186 186 + add(str(entity_id)) 187 187 + 188 188 + return unmapped 189 189 + 190 190 + 191 191 + def _sanitize_stage_name(relative_path: str) -> str: 192 192 + return relative_path.replace("/", "__") + ".staged.json" 193 193 + 194 194 + 195 195 + def _stage_unmapped_entity( 196 196 + staged_dir: Path, 197 197 + facet_name: str, 198 198 + file_type: str, 199 199 + relative_path: str, 200 200 + entity_id: str, 201 201 + source_data: str, 202 202 + ) -> Path: 203 203 + target_path = staged_dir / facet_name / file_type / _sanitize_stage_name(relative_path) 204 204 + target_path.parent.mkdir(parents=True, exist_ok=True) 205 205 + payload = { 206 206 + "reason": "unmapped_entity", 207 207 + "source_entity_id": entity_id, 208 208 + "explanation": ( 209 209 + f"Entity '{entity_id}' has no mapping in entities/state.json id_map" 210 210 + ), 211 211 + "source_path": relative_path, 212 212 + "source_data": source_data, 213 213 + "staged_at": datetime.now(timezone.utc).isoformat(), 214 214 + } 215 215 + target_path.write_text( 216 216 + json.dumps(payload, indent=2, ensure_ascii=False) + "\n", 217 217 + encoding="utf-8", 218 218 + ) 219 219 + return target_path 220 220 + 221 221 + 222 222 + def _stage_facet_json_conflict( 223 223 + staged_dir: Path, 224 224 + facet_name: str, 225 225 + relative_path: str, 226 226 + source_content: Any, 227 227 + target_content: Any, 228 228 + ) -> Path: 229 229 + target_path = staged_dir / facet_name / "facet_json" / _sanitize_stage_name(relative_path) 230 230 + target_path.parent.mkdir(parents=True, exist_ok=True) 231 231 + payload = { 232 232 + "reason": "facet_json_conflict", 233 233 + "source_content": source_content, 234 234 + "target_content": target_content, 235 235 + "staged_at": datetime.now(timezone.utc).isoformat(), 236 236 + } 237 237 + target_path.write_text( 238 238 + json.dumps(payload, indent=2, ensure_ascii=False) + "\n", 239 239 + encoding="utf-8", 240 240 + ) 241 241 + return target_path 242 242 + 243 243 + 244 244 + def _write_bytes(path: Path, raw_bytes: bytes) -> None: 245 245 + path.parent.mkdir(parents=True, exist_ok=True) 246 246 + path.write_bytes(raw_bytes) 247 247 + 248 248 + 249 249 + def _merge_facet_json( 250 250 + target_path: Path, 251 251 + raw_bytes: bytes, 252 252 + *, 253 253 + new_facet: bool, 254 254 + staged_dir: Path, 255 255 + facet_name: str, 256 256 + relative_path: str, 257 257 + ) -> dict[str, Any]: 258 258 + source_content = _parse_json_bytes(raw_bytes) 259 259 + if not target_path.exists() or new_facet: 260 260 + _write_bytes(target_path, raw_bytes) 261 261 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 262 262 + 263 263 + target_content = json.loads(target_path.read_text(encoding="utf-8")) 264 264 + if target_content == source_content: 265 265 + return {"status": "skipped", "reason": "facet_json_match"} 266 266 + 267 267 + staged_path = _stage_facet_json_conflict( 268 268 + staged_dir, facet_name, relative_path, source_content, target_content 269 269 + ) 270 270 + return { 271 271 + "status": "staged", 272 272 + "reason": "facet_json_conflict", 273 273 + "staged_path": str(staged_path), 274 274 + } 275 275 + 276 276 + 277 277 + def _merge_entity_relationship( 278 278 + facet_name: str, 279 279 + entity_id: str, 280 280 + raw_bytes: bytes, 281 281 + *, 282 282 + new_facet: bool, 283 283 + ) -> dict[str, Any]: 284 284 + source_relationship = _parse_json_bytes(raw_bytes) 285 285 + if not isinstance(source_relationship, dict): 286 286 + raise ValueError("entity_relationship content must be a JSON object") 287 287 + source_relationship["entity_id"] = entity_id 288 288 + 289 289 + target_relationship = {} 290 290 + if not new_facet: 291 291 + loaded = load_facet_relationship(facet_name, entity_id) 292 292 + if loaded is not None: 293 293 + target_relationship = loaded 294 294 + 295 295 + merged_relationship = {**source_relationship, **target_relationship} 296 296 + save_facet_relationship(facet_name, entity_id, merged_relationship) 297 297 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 298 298 + 299 299 + 300 300 + def _merge_observations( 301 301 + facet_name: str, 302 302 + entity_id: str, 303 303 + raw_bytes: bytes, 304 304 + *, 305 305 + new_facet: bool, 306 306 + ) -> dict[str, Any]: 307 307 + source_observations = _parse_jsonl_bytes(raw_bytes) 308 308 + target_observations = [] if new_facet else load_observations(facet_name, entity_id) 309 309 + seen = { 310 310 + (item.get("content", ""), item.get("observed_at")) for item in target_observations 311 311 + } 312 312 + merged_observations = list(target_observations) 313 313 + for item in source_observations: 314 314 + key = (item.get("content", ""), item.get("observed_at")) 315 315 + if key in seen: 316 316 + continue 317 317 + seen.add(key) 318 318 + merged_observations.append(item) 319 319 + 320 320 + save_observations(facet_name, entity_id, merged_observations) 321 321 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 322 322 + 323 323 + 324 324 + def _merge_detected_entities( 325 325 + target_path: Path, 326 326 + raw_bytes: bytes, 327 327 + *, 328 328 + new_facet: bool, 329 329 + ) -> dict[str, Any]: 330 330 + source_items = _parse_jsonl_bytes(raw_bytes) 331 331 + target_items = [] if new_facet else _read_jsonl(target_path) 332 332 + seen_ids = {item.get("id") for item in target_items if item.get("id")} 333 333 + new_items = [] 334 334 + for item in source_items: 335 335 + item_id = item.get("id", "") 336 336 + if item_id in seen_ids: 337 337 + continue 338 338 + new_items.append(item) 339 339 + _append_jsonl(target_path, new_items) 340 340 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 341 341 + 342 342 + 343 343 + def _merge_activity_config( 344 344 + target_path: Path, 345 345 + raw_bytes: bytes, 346 346 + *, 347 347 + new_facet: bool, 348 348 + ) -> dict[str, Any]: 349 349 + source_items = _parse_jsonl_bytes(raw_bytes) 350 350 + target_items = [] if new_facet else _read_jsonl(target_path) 351 351 + existing_ids = {item.get("id") for item in target_items} 352 352 + new_items = [item for item in source_items if item.get("id") not in existing_ids] 353 353 + _append_jsonl(target_path, new_items) 354 354 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 355 355 + 356 356 + 357 357 + def _merge_activity_records( 358 358 + target_path: Path, 359 359 + raw_bytes: bytes, 360 360 + *, 361 361 + new_facet: bool, 362 362 + ) -> dict[str, Any]: 363 363 + source_items = _parse_jsonl_bytes(raw_bytes) 364 364 + target_items = [] if new_facet else _read_jsonl(target_path) 365 365 + existing_ids = {item.get("id") for item in target_items} 366 366 + new_items = [item for item in source_items if item.get("id") not in existing_ids] 367 367 + _append_jsonl(target_path, new_items) 368 368 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 369 369 + 370 370 + 371 371 + def _merge_activity_output( 372 372 + target_path: Path, 373 373 + raw_bytes: bytes, 374 374 + output_dir: Path, 375 375 + *, 376 376 + new_facet: bool, 377 377 + ) -> dict[str, Any]: 378 378 + if output_dir.exists(): 379 379 + return {"status": "skipped", "reason": "output_dir_exists"} 380 380 + _write_bytes(target_path, raw_bytes) 381 381 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 382 382 + 383 383 + 384 384 + def _merge_todos( 385 385 + target_path: Path, 386 386 + raw_bytes: bytes, 387 387 + *, 388 388 + new_facet: bool, 389 389 + ) -> dict[str, Any]: 390 390 + source_items = _parse_jsonl_bytes(raw_bytes) 391 391 + target_items = [] if new_facet else _read_jsonl(target_path) 392 392 + seen = {(item["text"], item.get("created_at")) for item in target_items} 393 393 + new_items = [ 394 394 + item 395 395 + for item in source_items 396 396 + if (item["text"], item.get("created_at")) not in seen 397 397 + ] 398 398 + _append_jsonl(target_path, new_items) 399 399 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 400 400 + 401 401 + 402 402 + def _merge_calendar( 403 403 + target_path: Path, 404 404 + raw_bytes: bytes, 405 405 + *, 406 406 + new_facet: bool, 407 407 + ) -> dict[str, Any]: 408 408 + source_items = _parse_jsonl_bytes(raw_bytes) 409 409 + target_items = [] if new_facet else _read_jsonl(target_path) 410 410 + seen = {(item["title"], item.get("start")) for item in target_items} 411 411 + new_items = [ 412 412 + item 413 413 + for item in source_items 414 414 + if (item["title"], item.get("start")) not in seen 415 415 + ] 416 416 + _append_jsonl(target_path, new_items) 417 417 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 418 418 + 419 419 + 420 420 + def _merge_news( 421 421 + target_path: Path, 422 422 + raw_bytes: bytes, 423 423 + *, 424 424 + new_facet: bool, 425 425 + ) -> dict[str, Any]: 426 426 + if target_path.exists(): 427 427 + return {"status": "skipped", "reason": "news_exists"} 428 428 + _write_bytes(target_path, raw_bytes) 429 429 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 430 430 + 431 431 + 432 432 + def _merge_logs( 433 433 + target_path: Path, 434 434 + raw_bytes: bytes, 435 435 + *, 436 436 + new_facet: bool, 437 437 + ) -> dict[str, Any]: 438 438 + source_items = _parse_jsonl_bytes(raw_bytes) 439 439 + _append_jsonl(target_path, source_items) 440 440 + return {"status": "written", "reason": "new_facet" if new_facet else "overlap_merged"} 441 441 + 442 442 + 443 443 + def _remap_entity_ids( 444 444 + data: Any, 445 445 + id_map: dict[str, str], 446 446 + file_type: str, 447 447 + path_info: dict[str, str], 448 448 + ) -> tuple[Any, dict[str, str]]: 449 449 + updated_path_info = dict(path_info) 450 450 + 451 451 + if file_type == "entity_relationship": 452 452 + entity_id = path_info["entity_id"] 453 453 + mapped_id = _remap_entity_id(entity_id, id_map) 454 454 + if mapped_id is None: 455 455 + raise ValueError(f"Unmapped entity id: {entity_id}") 456 456 + if not isinstance(data, dict): 457 457 + raise ValueError("entity_relationship content must be a JSON object") 458 458 + updated_path_info["entity_id"] = mapped_id 459 459 + remapped = dict(data) 460 460 + remapped["entity_id"] = mapped_id 461 461 + return remapped, updated_path_info 462 462 + 463 463 + if file_type == "entity_observations": 464 464 + entity_id = path_info["entity_id"] 465 465 + mapped_id = _remap_entity_id(entity_id, id_map) 466 466 + if mapped_id is None: 467 467 + raise ValueError(f"Unmapped entity id: {entity_id}") 468 468 + updated_path_info["entity_id"] = mapped_id 469 469 + return data, updated_path_info 470 470 + 471 471 + if file_type == "detected_entities": 472 472 + remapped_items = [] 473 473 + for item in data: 474 474 + updated = dict(item) 475 475 + entity_id = updated.get("id") 476 476 + if entity_id: 477 477 + mapped_id = _remap_entity_id(str(entity_id), id_map) 478 478 + if mapped_id is None: 479 479 + raise ValueError(f"Unmapped entity id: {entity_id}") 480 480 + updated["id"] = mapped_id 481 481 + remapped_items.append(updated) 482 482 + return remapped_items, updated_path_info 483 483 + 484 484 + if file_type == "activity_records": 485 485 + remapped_items = [] 486 486 + for item in data: 487 487 + updated = dict(item) 488 488 + active_entities = updated.get("active_entities") 489 489 + if isinstance(active_entities, list): 490 490 + remapped_entities = [] 491 491 + for entity_id in active_entities: 492 492 + mapped_id = _remap_entity_id(str(entity_id), id_map) 493 493 + if mapped_id is None: 494 494 + raise ValueError(f"Unmapped entity id: {entity_id}") 495 495 + remapped_entities.append(mapped_id) 496 496 + updated["active_entities"] = remapped_entities 497 497 + remapped_items.append(updated) 498 498 + return remapped_items, updated_path_info 499 499 + 500 500 + return data, updated_path_info 501 501 + 502 502 + 503 503 + def _serialize_jsonl(items: list[dict[str, Any]]) -> bytes: 504 504 + if not items: 505 505 + return b"" 506 506 + return "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in items).encode( 507 507 + "utf-8" 508 508 + ) 509 509 + 510 510 + 511 511 + def process_facet( 512 512 + facet_name: str, 513 513 + files: list[dict], 514 514 + file_data: list[bytes], 515 515 + journal_root: Path, 516 516 + id_map: dict[str, str], 517 517 + log_path: Path, 518 518 + staged_dir: Path, 519 519 + received: dict[str, str], 520 520 + ) -> dict: 521 521 + if not _FACET_NAME_RE.match(facet_name): 522 522 + raise ValueError("Invalid facet name") 523 523 + 524 524 + facet_dir = journal_root / "facets" / facet_name 525 525 + new_facet = not facet_dir.exists() 526 526 + 527 527 + result = { 528 528 + "created": 0, 529 529 + "merged": 0, 530 530 + "skipped": 0, 531 531 + "staged": 0, 532 532 + "errors": [], 533 533 + "wrote_files": False, 534 534 + } 535 535 + 536 536 + for metadata, raw_bytes in zip(files, file_data, strict=True): 537 537 + raw_path = str(metadata.get("path", "")).strip() 538 538 + file_type = str(metadata.get("type", "")).strip() 539 539 + 540 540 + try: 541 541 + normalized_path, path_info = _parse_path(raw_path, file_type) 542 542 + relative_path = normalized_path.as_posix() 543 543 + item_id = f"{facet_name}/{relative_path}" 544 544 + content_hash = hashlib.sha256(raw_bytes).hexdigest() 545 545 + 546 546 + if received.get(item_id) == content_hash: 547 547 + result["skipped"] += 1 548 548 + _append_decision( 549 549 + log_path, 550 550 + { 551 551 + "ts": datetime.now(timezone.utc).isoformat(), 552 552 + "action": "facet_file_skipped", 553 553 + "item_type": file_type, 554 554 + "item_id": item_id, 555 555 + "facet": facet_name, 556 556 + "reason": "idempotent", 557 557 + }, 558 558 + ) 559 559 + continue 560 560 + 561 561 + parsed_data: Any = raw_bytes 562 562 + if file_type == "facet_json": 563 563 + parsed_data = _parse_json_bytes(raw_bytes) 564 564 + elif file_type in { 565 565 + "entity_observations", 566 566 + "detected_entities", 567 567 + "activity_config", 568 568 + "activity_records", 569 569 + "todos", 570 570 + "calendar", 571 571 + "logs", 572 572 + }: 573 573 + parsed_data = _parse_jsonl_bytes(raw_bytes) 574 574 + elif file_type == "entity_relationship": 575 575 + parsed_data = _parse_json_bytes(raw_bytes) 576 576 + 577 577 + if file_type in _ENTITY_FILE_TYPES: 578 578 + unmapped = _check_unmapped_entities(parsed_data, id_map, file_type, path_info) 579 579 + if unmapped: 580 580 + staged_path = _stage_unmapped_entity( 581 581 + staged_dir, 582 582 + facet_name, 583 583 + file_type, 584 584 + relative_path, 585 585 + unmapped[0], 586 586 + _decode_text(raw_bytes), 587 587 + ) 588 588 + result["staged"] += 1 589 589 + _append_decision( 590 590 + log_path, 591 591 + { 592 592 + "ts": datetime.now(timezone.utc).isoformat(), 593 593 + "action": "facet_file_staged", 594 594 + "item_type": file_type, 595 595 + "item_id": item_id, 596 596 + "facet": facet_name, 597 597 + "reason": "unmapped_entity", 598 598 + "staged_path": str(staged_path), 599 599 + }, 600 600 + ) 601 601 + continue 602 602 + 603 603 + parsed_data, path_info = _remap_entity_ids( 604 604 + parsed_data, id_map, file_type, path_info 605 605 + ) 606 606 + if file_type == "entity_relationship": 607 607 + raw_bytes = ( 608 608 + json.dumps(parsed_data, ensure_ascii=False, indent=2) + "\n" 609 609 + ).encode("utf-8") 610 610 + elif file_type in {"detected_entities", "activity_records"}: 611 611 + raw_bytes = _serialize_jsonl(parsed_data) 612 612 + 613 613 + target_path = facet_dir / normalized_path 614 614 + 615 615 + if file_type == "facet_json": 616 616 + merge_result = _merge_facet_json( 617 617 + target_path, 618 618 + raw_bytes, 619 619 + new_facet=new_facet, 620 620 + staged_dir=staged_dir, 621 621 + facet_name=facet_name, 622 622 + relative_path=relative_path, 623 623 + ) 624 624 + elif file_type == "entity_relationship": 625 625 + merge_result = _merge_entity_relationship( 626 626 + facet_name, 627 627 + path_info["entity_id"], 628 628 + raw_bytes, 629 629 + new_facet=new_facet, 630 630 + ) 631 631 + elif file_type == "entity_observations": 632 632 + merge_result = _merge_observations( 633 633 + facet_name, 634 634 + path_info["entity_id"], 635 635 + raw_bytes, 636 636 + new_facet=new_facet, 637 637 + ) 638 638 + elif file_type == "detected_entities": 639 639 + merge_result = _merge_detected_entities( 640 640 + target_path, raw_bytes, new_facet=new_facet 641 641 + ) 642 642 + elif file_type == "activity_config": 643 643 + merge_result = _merge_activity_config( 644 644 + target_path, raw_bytes, new_facet=new_facet 645 645 + ) 646 646 + elif file_type == "activity_records": 647 647 + merge_result = _merge_activity_records( 648 648 + target_path, raw_bytes, new_facet=new_facet 649 649 + ) 650 650 + elif file_type == "activity_output": 651 651 + output_dir = facet_dir / PurePosixPath(path_info["output_dir"]) 652 652 + merge_result = _merge_activity_output( 653 653 + target_path, 654 654 + raw_bytes, 655 655 + output_dir, 656 656 + new_facet=new_facet, 657 657 + ) 658 658 + elif file_type == "todos": 659 659 + merge_result = _merge_todos(target_path, raw_bytes, new_facet=new_facet) 660 660 + elif file_type == "calendar": 661 661 + merge_result = _merge_calendar(target_path, raw_bytes, new_facet=new_facet) 662 662 + elif file_type == "news": 663 663 + merge_result = _merge_news(target_path, raw_bytes, new_facet=new_facet) 664 664 + elif file_type == "logs": 665 665 + merge_result = _merge_logs(target_path, raw_bytes, new_facet=new_facet) 666 666 + else: 667 667 + raise ValueError(f"Unsupported file type: {file_type}") 668 668 + 669 669 + status = merge_result["status"] 670 670 + if status == "written": 671 671 + received[item_id] = content_hash 672 672 + bucket = "created" if new_facet else "merged" 673 673 + result[bucket] += 1 674 674 + result["wrote_files"] = True 675 675 + action = "facet_file_created" if new_facet else "facet_file_merged" 676 676 + elif status == "staged": 677 677 + result["staged"] += 1 678 678 + action = "facet_file_staged" 679 679 + else: 680 680 + received[item_id] = content_hash 681 681 + result["skipped"] += 1 682 682 + action = "facet_file_skipped" 683 683 + 684 684 + entry = { 685 685 + "ts": datetime.now(timezone.utc).isoformat(), 686 686 + "action": action, 687 687 + "item_type": file_type, 688 688 + "item_id": item_id, 689 689 + "facet": facet_name, 690 690 + "reason": merge_result["reason"], 691 691 + } 692 692 + if "staged_path" in merge_result: 693 693 + entry["staged_path"] = merge_result["staged_path"] 694 694 + _append_decision(log_path, entry) 695 695 + except Exception as exc: 696 696 + _append_decision( 697 697 + log_path, 698 698 + { 699 699 + "ts": datetime.now(timezone.utc).isoformat(), 700 700 + "action": "facet_file_error", 701 701 + "item_type": file_type, 702 702 + "item_id": f"{facet_name}/{raw_path}", 703 703 + "facet": facet_name, 704 704 + "reason": str(exc), 705 705 + }, 706 706 + ) 707 707 + result["errors"].append( 708 708 + { 709 709 + "facet": facet_name, 710 710 + "path": raw_path, 711 711 + "error": str(exc), 712 712 + } 713 713 + ) 714 714 + 715 715 + return result 716 716 + 717 717 + 718 718 + __all__ = ["process_facet"]

+141

apps/import/ingest.py

reviewed

··· 42 42 _DAY_RE = re.compile(r"^\d{8}$") 43 43 _SEGMENT_RE = re.compile(r"^\d{6}_\d+$") 44 44 _STREAM_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$") 45 45 + _FACET_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*$") 45 46 46 47 47 48 def _append_decision(log_path: Path, entry: dict) -> None: ··· 60 61 except Exception: 61 62 Path(tmp_path).unlink(missing_ok=True) 62 63 raise 64 64 + 65 65 + 66 66 + from .facet_ingest import process_facet 63 67 64 68 65 69 def register_ingest_routes(bp) -> None: ··· 560 564 "errors": errors, 561 565 } 562 566 ) 567 567 + 568 568 + @bp.route("/journal/<key_prefix>/ingest/facets", methods=["POST"]) 569 569 + @require_journal_source 570 570 + def ingest_facets(key_prefix: str): 571 571 + if g.journal_source["key"][:8] != key_prefix: 572 572 + abort(403, description="Key prefix mismatch") 573 573 + 574 574 + metadata_raw = request.form.get("metadata") 575 575 + if not metadata_raw: 576 576 + return jsonify({"error": "Missing metadata"}), 400 577 577 + 578 578 + try: 579 579 + metadata = json.loads(metadata_raw) 580 580 + except json.JSONDecodeError: 581 581 + return jsonify({"error": "Invalid metadata JSON"}), 400 582 582 + 583 583 + if not isinstance(metadata, dict): 584 584 + return jsonify({"error": "Invalid metadata JSON"}), 400 585 585 + 586 586 + facets = metadata.get("facets") 587 587 + if not isinstance(facets, list): 588 588 + return jsonify({"error": "Missing facets array"}), 400 589 589 + 590 590 + state_dir = get_state_directory(key_prefix) 591 591 + entities_state_path = state_dir / "entities" / "state.json" 592 592 + facets_state_path = state_dir / "facets" / "state.json" 593 593 + log_path = state_dir / "facets" / "log.jsonl" 594 594 + staged_dir = state_dir / "facets" / "staged" 595 595 + 596 596 + try: 597 597 + entities_state = json.loads(entities_state_path.read_text(encoding="utf-8")) 598 598 + except (OSError, json.JSONDecodeError): 599 599 + entities_state = {} 600 600 + if not isinstance(entities_state, dict): 601 601 + entities_state = {} 602 602 + id_map = entities_state.get("id_map") 603 603 + if not isinstance(id_map, dict): 604 604 + id_map = {} 605 605 + 606 606 + try: 607 607 + facets_state = json.loads(facets_state_path.read_text(encoding="utf-8")) 608 608 + except (OSError, json.JSONDecodeError): 609 609 + facets_state = {} 610 610 + if not isinstance(facets_state, dict): 611 611 + facets_state = {} 612 612 + received = facets_state.get("received") 613 613 + if not isinstance(received, dict): 614 614 + received = {} 615 615 + facets_state = {"received": dict(received)} 616 616 + 617 617 + created = 0 618 618 + merged = 0 619 619 + skipped = 0 620 620 + staged = 0 621 621 + errors: list[dict[str, str]] = [] 622 622 + written_facets: set[str] = set() 623 623 + 624 624 + for facet_idx, facet in enumerate(facets): 625 625 + if not isinstance(facet, dict): 626 626 + return jsonify({"error": "Facet metadata must be an object"}), 400 627 627 + 628 628 + facet_name = str(facet.get("name", "")).strip() 629 629 + files = facet.get("files") 630 630 + if not facet_name: 631 631 + return jsonify({"error": "Facet name is required"}), 400 632 632 + if not _FACET_NAME_RE.match(facet_name): 633 633 + return jsonify({"error": "Invalid facet name"}), 400 634 634 + if not isinstance(files, list): 635 635 + return jsonify({"error": "Facet files must be an array"}), 400 636 636 + 637 637 + file_bytes: list[bytes] = [] 638 638 + normalized_files: list[dict[str, str]] = [] 639 639 + for file_idx, file_meta in enumerate(files): 640 640 + if not isinstance(file_meta, dict): 641 641 + return jsonify({"error": "Facet file metadata must be an object"}), 400 642 642 + 643 643 + path_value = file_meta.get("path") 644 644 + type_value = file_meta.get("type") 645 645 + if not isinstance(path_value, str) or not isinstance(type_value, str): 646 646 + return ( 647 647 + jsonify({"error": "Facet file metadata must include path and type"}), 648 648 + 400, 649 649 + ) 650 650 + 651 651 + upload = request.files.get(f"files_{facet_idx}_{file_idx}") 652 652 + if upload is None: 653 653 + return ( 654 654 + jsonify( 655 655 + { 656 656 + "error": ( 657 657 + f"Missing uploaded file for facet {facet_idx} file {file_idx}" 658 658 + ) 659 659 + } 660 660 + ), 661 661 + 400, 662 662 + ) 663 663 + 664 664 + file_bytes.append(upload.read()) 665 665 + normalized_files.append({"path": path_value, "type": type_value}) 666 666 + 667 667 + facet_result = process_facet( 668 668 + facet_name=facet_name, 669 669 + files=normalized_files, 670 670 + file_data=file_bytes, 671 671 + journal_root=Path(state.journal_root), 672 672 + id_map=id_map, 673 673 + log_path=log_path, 674 674 + staged_dir=staged_dir, 675 675 + received=facets_state["received"], 676 676 + ) 677 677 + created += facet_result["created"] 678 678 + merged += facet_result["merged"] 679 679 + skipped += facet_result["skipped"] 680 680 + staged += facet_result["staged"] 681 681 + errors.extend(facet_result["errors"]) 682 682 + if facet_result["wrote_files"]: 683 683 + written_facets.add(facet_name) 684 684 + 685 685 + _write_state_atomic(facets_state_path, facets_state) 686 686 + 687 687 + if written_facets: 688 688 + source = g.journal_source 689 689 + source.setdefault("stats", {}) 690 690 + source["stats"]["facets_received"] = ( 691 691 + source["stats"].get("facets_received", 0) + len(written_facets) 692 692 + ) 693 693 + save_journal_source(source) 694 694 + 695 695 + return jsonify( 696 696 + { 697 697 + "created": created, 698 698 + "merged": merged, 699 699 + "skipped": skipped, 700 700 + "staged": staged, 701 701 + "errors": errors, 702 702 + } 703 703 + )

+1124

tests/test_facet_ingest.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + import hashlib 7 7 + import json 8 8 + from importlib import import_module 9 9 + from io import BytesIO 10 10 + from pathlib import Path 11 11 + 12 12 + import pytest 13 13 + from flask import Blueprint, Flask 14 14 + 15 15 + import convey.state 16 16 + 17 17 + journal_sources = import_module("apps.import.journal_sources") 18 18 + ingest = import_module("apps.import.ingest") 19 19 + 20 20 + create_state_directory = journal_sources.create_state_directory 21 21 + generate_key = journal_sources.generate_key 22 22 + get_state_directory = journal_sources.get_state_directory 23 23 + load_journal_source = journal_sources.load_journal_source 24 24 + save_journal_source = journal_sources.save_journal_source 25 25 + register_ingest_routes = ingest.register_ingest_routes 26 26 + 27 27 + 28 28 + @pytest.fixture 29 29 + def journal_env(tmp_path, monkeypatch): 30 30 + monkeypatch.setattr(convey.state, "journal_root", str(tmp_path), raising=False) 31 31 + monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path)) 32 32 + (tmp_path / "apps" / "import" / "journal_sources").mkdir( 33 33 + parents=True, exist_ok=True 34 34 + ) 35 35 + return tmp_path 36 36 + 37 37 + 38 38 + def _source(name="test-source", key=None, **overrides): 39 39 + if key is None: 40 40 + key = generate_key() 41 41 + source = { 42 42 + "name": name, 43 43 + "key": key, 44 44 + "created_at": 1000, 45 45 + "enabled": True, 46 46 + "revoked": False, 47 47 + "revoked_at": None, 48 48 + "stats": { 49 49 + "segments_received": 0, 50 50 + "entities_received": 0, 51 51 + "facets_received": 0, 52 52 + "imports_received": 0, 53 53 + "config_received": 0, 54 54 + }, 55 55 + } 56 56 + source.update(overrides) 57 57 + return source 58 58 + 59 59 + 60 60 + @pytest.fixture 61 61 + def ingest_env(journal_env): 62 62 + key = generate_key() 63 63 + source = _source(key=key) 64 64 + save_journal_source(source) 65 65 + key_prefix = key[:8] 66 66 + create_state_directory(journal_env, key_prefix) 67 67 + 68 68 + entity_state = { 69 69 + "id_map": { 70 70 + "source_entity": "target_entity", 71 71 + "same_entity": "same_entity", 72 72 + }, 73 73 + "received": {}, 74 74 + } 75 75 + ( 76 76 + get_state_directory(key_prefix) / "entities" / "state.json" 77 77 + ).write_text(json.dumps(entity_state, indent=2), encoding="utf-8") 78 78 + 79 79 + app = Flask(__name__) 80 80 + app.config["TESTING"] = True 81 81 + bp = Blueprint("import-test", __name__, url_prefix="/app/import") 82 82 + register_ingest_routes(bp) 83 83 + app.register_blueprint(bp) 84 84 + 85 85 + return { 86 86 + "root": journal_env, 87 87 + "key": key, 88 88 + "key_prefix": key_prefix, 89 89 + "source": source, 90 90 + "client": app.test_client(), 91 91 + } 92 92 + 93 93 + 94 94 + def _post_facets(client, key, key_prefix, facets_metadata, file_map): 95 95 + data = {"metadata": json.dumps({"facets": facets_metadata})} 96 96 + data.update(file_map) 97 97 + return client.post( 98 98 + f"/app/import/journal/{key_prefix}/ingest/facets", 99 99 + headers={"Authorization": f"Bearer {key}"}, 100 100 + data=data, 101 101 + content_type="multipart/form-data", 102 102 + ) 103 103 + 104 104 + 105 105 + def _build_request(facets): 106 106 + facets_metadata = [] 107 107 + file_map = {} 108 108 + for facet_idx, facet in enumerate(facets): 109 109 + facet_meta = {"name": facet["name"], "files": []} 110 110 + for file_idx, file_info in enumerate(facet["files"]): 111 111 + facet_meta["files"].append( 112 112 + {"path": file_info["path"], "type": file_info["type"]} 113 113 + ) 114 114 + filename = Path(file_info["path"]).name 115 115 + file_map[f"files_{facet_idx}_{file_idx}"] = ( 116 116 + BytesIO(file_info["content"]), 117 117 + filename, 118 118 + ) 119 119 + facets_metadata.append(facet_meta) 120 120 + return facets_metadata, file_map 121 121 + 122 122 + 123 123 + def _read_state(key_prefix: str) -> dict: 124 124 + state_path = get_state_directory(key_prefix) / "facets" / "state.json" 125 125 + return json.loads(state_path.read_text(encoding="utf-8")) 126 126 + 127 127 + 128 128 + def _read_log(key_prefix: str) -> list[dict]: 129 129 + log_path = get_state_directory(key_prefix) / "facets" / "log.jsonl" 130 130 + if not log_path.exists(): 131 131 + return [] 132 132 + return [ 133 133 + json.loads(line) 134 134 + for line in log_path.read_text(encoding="utf-8").splitlines() 135 135 + if line.strip() 136 136 + ] 137 137 + 138 138 + 139 139 + def _read_staged(key_prefix: str, facet: str, file_type: str, relative_path: str) -> dict: 140 140 + staged_name = relative_path.replace("/", "__") + ".staged.json" 141 141 + staged_path = ( 142 142 + get_state_directory(key_prefix) 143 143 + / "facets" 144 144 + / "staged" 145 145 + / facet 146 146 + / file_type 147 147 + / staged_name 148 148 + ) 149 149 + return json.loads(staged_path.read_text(encoding="utf-8")) 150 150 + 151 151 + 152 152 + def _json_bytes(data: dict) -> bytes: 153 153 + return (json.dumps(data, ensure_ascii=False, indent=2) + "\n").encode("utf-8") 154 154 + 155 155 + 156 156 + def _jsonl_bytes(items: list[dict]) -> bytes: 157 157 + return "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in items).encode( 158 158 + "utf-8" 159 159 + ) 160 160 + 161 161 + 162 162 + def _read_json(path: Path) -> dict: 163 163 + return json.loads(path.read_text(encoding="utf-8")) 164 164 + 165 165 + 166 166 + def _read_jsonl_file(path: Path) -> list[dict]: 167 167 + if not path.exists(): 168 168 + return [] 169 169 + return [ 170 170 + json.loads(line) 171 171 + for line in path.read_text(encoding="utf-8").splitlines() 172 172 + if line.strip() 173 173 + ] 174 174 + 175 175 + 176 176 + def _hash_bytes(content: bytes) -> str: 177 177 + return hashlib.sha256(content).hexdigest() 178 178 + 179 179 + 180 180 + def _write_json(path: Path, data: dict) -> None: 181 181 + path.parent.mkdir(parents=True, exist_ok=True) 182 182 + path.write_text( 183 183 + json.dumps(data, ensure_ascii=False, indent=2) + "\n", 184 184 + encoding="utf-8", 185 185 + ) 186 186 + 187 187 + 188 188 + def _write_jsonl(path: Path, items: list[dict]) -> None: 189 189 + path.parent.mkdir(parents=True, exist_ok=True) 190 190 + path.write_text(_jsonl_bytes(items).decode("utf-8"), encoding="utf-8") 191 191 + 192 192 + 193 193 + def test_auth_missing(ingest_env): 194 194 + env = ingest_env 195 195 + response = env["client"].post( 196 196 + f"/app/import/journal/{env['key_prefix']}/ingest/facets", 197 197 + data={"metadata": json.dumps({"facets": []})}, 198 198 + content_type="multipart/form-data", 199 199 + ) 200 200 + assert response.status_code == 401 201 201 + 202 202 + 203 203 + def test_auth_invalid(ingest_env): 204 204 + env = ingest_env 205 205 + response = env["client"].post( 206 206 + f"/app/import/journal/{env['key_prefix']}/ingest/facets", 207 207 + headers={"Authorization": "Bearer wrong-token"}, 208 208 + data={"metadata": json.dumps({"facets": []})}, 209 209 + content_type="multipart/form-data", 210 210 + ) 211 211 + assert response.status_code == 401 212 212 + 213 213 + 214 214 + def test_auth_revoked(ingest_env): 215 215 + env = ingest_env 216 216 + env["source"]["revoked"] = True 217 217 + env["source"]["revoked_at"] = 12345 218 218 + save_journal_source(env["source"]) 219 219 + 220 220 + response = env["client"].post( 221 221 + f"/app/import/journal/{env['key_prefix']}/ingest/facets", 222 222 + headers={"Authorization": f"Bearer {env['key']}"}, 223 223 + data={"metadata": json.dumps({"facets": []})}, 224 224 + content_type="multipart/form-data", 225 225 + ) 226 226 + assert response.status_code == 403 227 227 + 228 228 + 229 229 + def test_key_prefix_mismatch(ingest_env): 230 230 + env = ingest_env 231 231 + response = env["client"].post( 232 232 + "/app/import/journal/deadbeef/ingest/facets", 233 233 + headers={"Authorization": f"Bearer {env['key']}"}, 234 234 + data={"metadata": json.dumps({"facets": []})}, 235 235 + content_type="multipart/form-data", 236 236 + ) 237 237 + assert response.status_code == 403 238 238 + 239 239 + 240 240 + def test_missing_metadata(ingest_env): 241 241 + env = ingest_env 242 242 + response = env["client"].post( 243 243 + f"/app/import/journal/{env['key_prefix']}/ingest/facets", 244 244 + headers={"Authorization": f"Bearer {env['key']}"}, 245 245 + content_type="multipart/form-data", 246 246 + ) 247 247 + assert response.status_code == 400 248 248 + assert response.get_json() == {"error": "Missing metadata"} 249 249 + 250 250 + 251 251 + def test_invalid_metadata_json(ingest_env): 252 252 + env = ingest_env 253 253 + response = env["client"].post( 254 254 + f"/app/import/journal/{env['key_prefix']}/ingest/facets", 255 255 + headers={"Authorization": f"Bearer {env['key']}"}, 256 256 + data={"metadata": "not-json"}, 257 257 + content_type="multipart/form-data", 258 258 + ) 259 259 + assert response.status_code == 400 260 260 + assert response.get_json() == {"error": "Invalid metadata JSON"} 261 261 + 262 262 + 263 263 + def test_unsafe_facet_name(ingest_env): 264 264 + env = ingest_env 265 265 + for bad_name in ["../etc", "foo/bar", ".", "..", "FOO"]: 266 266 + facets = [ 267 267 + { 268 268 + "name": bad_name, 269 269 + "files": [ 270 270 + { 271 271 + "path": "news/20260305.md", 272 272 + "type": "news", 273 273 + "content": b"x", 274 274 + } 275 275 + ], 276 276 + } 277 277 + ] 278 278 + metadata, file_map = _build_request(facets) 279 279 + response = _post_facets( 280 280 + env["client"], env["key"], env["key_prefix"], metadata, file_map 281 281 + ) 282 282 + assert response.status_code == 400, f"Expected 400 for facet name: {bad_name}" 283 283 + assert response.get_json() == {"error": "Invalid facet name"} 284 284 + 285 285 + 286 286 + def test_new_facet_all_types(ingest_env): 287 287 + env = ingest_env 288 288 + facets = [ 289 289 + { 290 290 + "name": "personal", 291 291 + "files": [ 292 292 + {"path": "facet.json", "type": "facet_json", "content": _json_bytes({"title": "Personal"})}, 293 293 + { 294 294 + "path": "entities/same_entity/entity.json", 295 295 + "type": "entity_relationship", 296 296 + "content": _json_bytes({"description": "Close contact", "attached_at": 100}), 297 297 + }, 298 298 + { 299 299 + "path": "entities/same_entity/observations.jsonl", 300 300 + "type": "entity_observations", 301 301 + "content": _jsonl_bytes( 302 302 + [{"content": "Likes tea", "observed_at": 1}] 303 303 + ), 304 304 + }, 305 305 + { 306 306 + "path": "entities/20260305.jsonl", 307 307 + "type": "detected_entities", 308 308 + "content": _jsonl_bytes( 309 309 + [{"id": "same_entity", "name": "Same Entity", "type": "Person"}] 310 310 + ), 311 311 + }, 312 312 + { 313 313 + "path": "activities/activities.jsonl", 314 314 + "type": "activity_config", 315 315 + "content": _jsonl_bytes( 316 316 + [{"id": "coding", "name": "Coding", "priority": "high"}] 317 317 + ), 318 318 + }, 319 319 + { 320 320 + "path": "activities/20260305.jsonl", 321 321 + "type": "activity_records", 322 322 + "content": _jsonl_bytes( 323 323 + [ 324 324 + { 325 325 + "id": "coding_093000_300", 326 326 + "activity": "coding", 327 327 + "active_entities": ["same_entity"], 328 328 + } 329 329 + ] 330 330 + ), 331 331 + }, 332 332 + { 333 333 + "path": "activities/20260305/coding_093000_300/session_review.md", 334 334 + "type": "activity_output", 335 335 + "content": b"# Session\n", 336 336 + }, 337 337 + { 338 338 + "path": "todos/20260305.jsonl", 339 339 + "type": "todos", 340 340 + "content": _jsonl_bytes([{"text": "Ship it", "created_at": 10}]), 341 341 + }, 342 342 + { 343 343 + "path": "calendar/20260305.jsonl", 344 344 + "type": "calendar", 345 345 + "content": _jsonl_bytes([{"title": "Standup", "start": "09:00"}]), 346 346 + }, 347 347 + { 348 348 + "path": "news/20260305.md", 349 349 + "type": "news", 350 350 + "content": b"# News\n", 351 351 + }, 352 352 + { 353 353 + "path": "logs/20260305.jsonl", 354 354 + "type": "logs", 355 355 + "content": _jsonl_bytes([{"event": "ingested"}]), 356 356 + }, 357 357 + ], 358 358 + } 359 359 + ] 360 360 + metadata, file_map = _build_request(facets) 361 361 + 362 362 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 363 363 + 364 364 + assert response.status_code == 200 365 365 + assert response.get_json() == { 366 366 + "created": 11, 367 367 + "merged": 0, 368 368 + "skipped": 0, 369 369 + "staged": 0, 370 370 + "errors": [], 371 371 + } 372 372 + 373 373 + facet_root = env["root"] / "facets" / "personal" 374 374 + assert _read_json(facet_root / "facet.json") == {"title": "Personal"} 375 375 + assert _read_json(facet_root / "entities" / "same_entity" / "entity.json") == { 376 376 + "description": "Close contact", 377 377 + "attached_at": 100, 378 378 + "entity_id": "same_entity", 379 379 + } 380 380 + assert _read_jsonl_file( 381 381 + facet_root / "entities" / "same_entity" / "observations.jsonl" 382 382 + ) == [{"content": "Likes tea", "observed_at": 1}] 383 383 + assert _read_jsonl_file(facet_root / "entities" / "20260305.jsonl")[0]["id"] == "same_entity" 384 384 + assert _read_jsonl_file(facet_root / "activities" / "activities.jsonl")[0]["id"] == "coding" 385 385 + assert _read_jsonl_file(facet_root / "activities" / "20260305.jsonl")[0]["id"] == "coding_093000_300" 386 386 + assert ( 387 387 + facet_root / "activities" / "20260305" / "coding_093000_300" / "session_review.md" 388 388 + ).read_text(encoding="utf-8") == "# Session\n" 389 389 + assert _read_jsonl_file(facet_root / "todos" / "20260305.jsonl")[0]["text"] == "Ship it" 390 390 + assert _read_jsonl_file(facet_root / "calendar" / "20260305.jsonl")[0]["title"] == "Standup" 391 391 + assert (facet_root / "news" / "20260305.md").read_text(encoding="utf-8") == "# News\n" 392 392 + assert _read_jsonl_file(facet_root / "logs" / "20260305.jsonl")[0]["event"] == "ingested" 393 393 + 394 394 + source = load_journal_source(env["key"]) 395 395 + assert source["stats"]["facets_received"] == 1 396 396 + 397 397 + 398 398 + def test_existing_facet_merge_entity_relationship(ingest_env): 399 399 + env = ingest_env 400 400 + target_path = env["root"] / "facets" / "work" / "entities" / "same_entity" / "entity.json" 401 401 + _write_json( 402 402 + target_path, 403 403 + {"entity_id": "same_entity", "description": "Keep target", "attached_at": 200}, 404 404 + ) 405 405 + 406 406 + facets = [ 407 407 + { 408 408 + "name": "work", 409 409 + "files": [ 410 410 + { 411 411 + "path": "entities/same_entity/entity.json", 412 412 + "type": "entity_relationship", 413 413 + "content": _json_bytes({"description": "Source desc", "last_seen": 999}), 414 414 + } 415 415 + ], 416 416 + } 417 417 + ] 418 418 + metadata, file_map = _build_request(facets) 419 419 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 420 420 + 421 421 + assert response.status_code == 200 422 422 + assert response.get_json()["merged"] == 1 423 423 + assert _read_json(target_path) == { 424 424 + "description": "Keep target", 425 425 + "last_seen": 999, 426 426 + "attached_at": 200, 427 427 + "entity_id": "same_entity", 428 428 + } 429 429 + 430 430 + 431 431 + def test_existing_facet_merge_observations(ingest_env): 432 432 + env = ingest_env 433 433 + target_path = ( 434 434 + env["root"] / "facets" / "work" / "entities" / "same_entity" / "observations.jsonl" 435 435 + ) 436 436 + _write_jsonl( 437 437 + target_path, 438 438 + [ 439 439 + {"content": "Likes tea", "observed_at": 1}, 440 440 + {"content": "Prefers email", "observed_at": 2}, 441 441 + ], 442 442 + ) 443 443 + 444 444 + facets = [ 445 445 + { 446 446 + "name": "work", 447 447 + "files": [ 448 448 + { 449 449 + "path": "entities/same_entity/observations.jsonl", 450 450 + "type": "entity_observations", 451 451 + "content": _jsonl_bytes( 452 452 + [ 453 453 + {"content": "Likes tea", "observed_at": 1}, 454 454 + {"content": "Uses vim", "observed_at": 3}, 455 455 + ] 456 456 + ), 457 457 + } 458 458 + ], 459 459 + } 460 460 + ] 461 461 + metadata, file_map = _build_request(facets) 462 462 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 463 463 + 464 464 + assert response.status_code == 200 465 465 + assert response.get_json()["merged"] == 1 466 466 + assert _read_jsonl_file(target_path) == [ 467 467 + {"content": "Likes tea", "observed_at": 1}, 468 468 + {"content": "Prefers email", "observed_at": 2}, 469 469 + {"content": "Uses vim", "observed_at": 3}, 470 470 + ] 471 471 + 472 472 + 473 473 + def test_existing_facet_merge_detected_entities(ingest_env): 474 474 + env = ingest_env 475 475 + target_path = env["root"] / "facets" / "work" / "entities" / "20260305.jsonl" 476 476 + _write_jsonl(target_path, [{"id": "same_entity", "name": "Same", "type": "Person"}]) 477 477 + 478 478 + facets = [ 479 479 + { 480 480 + "name": "work", 481 481 + "files": [ 482 482 + { 483 483 + "path": "entities/20260305.jsonl", 484 484 + "type": "detected_entities", 485 485 + "content": _jsonl_bytes( 486 486 + [ 487 487 + {"id": "same_entity", "name": "Same", "type": "Person"}, 488 488 + {"id": "source_entity", "name": "Source", "type": "Person"}, 489 489 + ] 490 490 + ), 491 491 + } 492 492 + ], 493 493 + } 494 494 + ] 495 495 + metadata, file_map = _build_request(facets) 496 496 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 497 497 + 498 498 + assert response.status_code == 200 499 499 + assert response.get_json()["merged"] == 1 500 500 + items = _read_jsonl_file(target_path) 501 501 + assert [item["id"] for item in items] == ["same_entity", "target_entity"] 502 502 + 503 503 + 504 504 + def test_existing_facet_merge_activity_config(ingest_env): 505 505 + env = ingest_env 506 506 + target_path = env["root"] / "facets" / "work" / "activities" / "activities.jsonl" 507 507 + _write_jsonl(target_path, [{"id": "coding", "name": "Coding"}]) 508 508 + 509 509 + facets = [ 510 510 + { 511 511 + "name": "work", 512 512 + "files": [ 513 513 + { 514 514 + "path": "activities/activities.jsonl", 515 515 + "type": "activity_config", 516 516 + "content": _jsonl_bytes( 517 517 + [ 518 518 + {"id": "coding", "name": "Coding"}, 519 519 + {"id": "meeting", "name": "Meeting"}, 520 520 + ] 521 521 + ), 522 522 + } 523 523 + ], 524 524 + } 525 525 + ] 526 526 + metadata, file_map = _build_request(facets) 527 527 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 528 528 + 529 529 + assert response.status_code == 200 530 530 + assert response.get_json()["merged"] == 1 531 531 + assert [item["id"] for item in _read_jsonl_file(target_path)] == ["coding", "meeting"] 532 532 + 533 533 + 534 534 + def test_existing_facet_merge_activity_records(ingest_env): 535 535 + env = ingest_env 536 536 + target_path = env["root"] / "facets" / "work" / "activities" / "20260305.jsonl" 537 537 + _write_jsonl( 538 538 + target_path, 539 539 + [{"id": "coding_1", "activity": "coding", "active_entities": ["same_entity"]}], 540 540 + ) 541 541 + 542 542 + facets = [ 543 543 + { 544 544 + "name": "work", 545 545 + "files": [ 546 546 + { 547 547 + "path": "activities/20260305.jsonl", 548 548 + "type": "activity_records", 549 549 + "content": _jsonl_bytes( 550 550 + [ 551 551 + {"id": "coding_1", "activity": "coding", "active_entities": ["same_entity"]}, 552 552 + {"id": "coding_2", "activity": "coding", "active_entities": ["source_entity"]}, 553 553 + ] 554 554 + ), 555 555 + } 556 556 + ], 557 557 + } 558 558 + ] 559 559 + metadata, file_map = _build_request(facets) 560 560 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 561 561 + 562 562 + assert response.status_code == 200 563 563 + assert response.get_json()["merged"] == 1 564 564 + items = _read_jsonl_file(target_path) 565 565 + assert [item["id"] for item in items] == ["coding_1", "coding_2"] 566 566 + assert items[1]["active_entities"] == ["target_entity"] 567 567 + 568 568 + 569 569 + def test_existing_facet_merge_activity_output_skip(ingest_env): 570 570 + env = ingest_env 571 571 + target_file = ( 572 572 + env["root"] 573 573 + / "facets" 574 574 + / "work" 575 575 + / "activities" 576 576 + / "20260305" 577 577 + / "coding_093000_300" 578 578 + / "session_review.md" 579 579 + ) 580 580 + target_file.parent.mkdir(parents=True, exist_ok=True) 581 581 + target_file.write_text("existing\n", encoding="utf-8") 582 582 + 583 583 + facets = [ 584 584 + { 585 585 + "name": "work", 586 586 + "files": [ 587 587 + { 588 588 + "path": "activities/20260305/coding_093000_300/session_review.md", 589 589 + "type": "activity_output", 590 590 + "content": b"new\n", 591 591 + } 592 592 + ], 593 593 + } 594 594 + ] 595 595 + metadata, file_map = _build_request(facets) 596 596 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 597 597 + 598 598 + assert response.status_code == 200 599 599 + assert response.get_json()["skipped"] == 1 600 600 + assert target_file.read_text(encoding="utf-8") == "existing\n" 601 601 + 602 602 + 603 603 + def test_existing_facet_merge_activity_output_copy(ingest_env): 604 604 + env = ingest_env 605 605 + (env["root"] / "facets" / "work").mkdir(parents=True, exist_ok=True) 606 606 + 607 607 + facets = [ 608 608 + { 609 609 + "name": "work", 610 610 + "files": [ 611 611 + { 612 612 + "path": "activities/20260305/coding_093000_300/session_review.md", 613 613 + "type": "activity_output", 614 614 + "content": b"copied\n", 615 615 + } 616 616 + ], 617 617 + } 618 618 + ] 619 619 + metadata, file_map = _build_request(facets) 620 620 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 621 621 + 622 622 + target_file = ( 623 623 + env["root"] 624 624 + / "facets" 625 625 + / "work" 626 626 + / "activities" 627 627 + / "20260305" 628 628 + / "coding_093000_300" 629 629 + / "session_review.md" 630 630 + ) 631 631 + assert response.status_code == 200 632 632 + assert response.get_json()["merged"] == 1 633 633 + assert target_file.read_text(encoding="utf-8") == "copied\n" 634 634 + 635 635 + 636 636 + def test_existing_facet_merge_todos(ingest_env): 637 637 + env = ingest_env 638 638 + target_path = env["root"] / "facets" / "work" / "todos" / "20260305.jsonl" 639 639 + _write_jsonl(target_path, [{"text": "Ship it", "created_at": 1}]) 640 640 + 641 641 + facets = [ 642 642 + { 643 643 + "name": "work", 644 644 + "files": [ 645 645 + { 646 646 + "path": "todos/20260305.jsonl", 647 647 + "type": "todos", 648 648 + "content": _jsonl_bytes( 649 649 + [ 650 650 + {"text": "Ship it", "created_at": 1}, 651 651 + {"text": "Review PR", "created_at": 2}, 652 652 + ] 653 653 + ), 654 654 + } 655 655 + ], 656 656 + } 657 657 + ] 658 658 + metadata, file_map = _build_request(facets) 659 659 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 660 660 + 661 661 + assert response.status_code == 200 662 662 + assert response.get_json()["merged"] == 1 663 663 + assert _read_jsonl_file(target_path) == [ 664 664 + {"text": "Ship it", "created_at": 1}, 665 665 + {"text": "Review PR", "created_at": 2}, 666 666 + ] 667 667 + 668 668 + 669 669 + def test_existing_facet_merge_calendar(ingest_env): 670 670 + env = ingest_env 671 671 + target_path = env["root"] / "facets" / "work" / "calendar" / "20260305.jsonl" 672 672 + _write_jsonl(target_path, [{"title": "Standup", "start": "09:00"}]) 673 673 + 674 674 + facets = [ 675 675 + { 676 676 + "name": "work", 677 677 + "files": [ 678 678 + { 679 679 + "path": "calendar/20260305.jsonl", 680 680 + "type": "calendar", 681 681 + "content": _jsonl_bytes( 682 682 + [ 683 683 + {"title": "Standup", "start": "09:00"}, 684 684 + {"title": "Demo", "start": "14:00"}, 685 685 + ] 686 686 + ), 687 687 + } 688 688 + ], 689 689 + } 690 690 + ] 691 691 + metadata, file_map = _build_request(facets) 692 692 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 693 693 + 694 694 + assert response.status_code == 200 695 695 + assert response.get_json()["merged"] == 1 696 696 + assert _read_jsonl_file(target_path) == [ 697 697 + {"title": "Standup", "start": "09:00"}, 698 698 + {"title": "Demo", "start": "14:00"}, 699 699 + ] 700 700 + 701 701 + 702 702 + def test_existing_facet_merge_news_skip(ingest_env): 703 703 + env = ingest_env 704 704 + target_path = env["root"] / "facets" / "work" / "news" / "20260305.md" 705 705 + target_path.parent.mkdir(parents=True, exist_ok=True) 706 706 + target_path.write_text("existing\n", encoding="utf-8") 707 707 + 708 708 + facets = [ 709 709 + { 710 710 + "name": "work", 711 711 + "files": [ 712 712 + { 713 713 + "path": "news/20260305.md", 714 714 + "type": "news", 715 715 + "content": b"new\n", 716 716 + } 717 717 + ], 718 718 + } 719 719 + ] 720 720 + metadata, file_map = _build_request(facets) 721 721 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 722 722 + 723 723 + assert response.status_code == 200 724 724 + assert response.get_json()["skipped"] == 1 725 725 + assert target_path.read_text(encoding="utf-8") == "existing\n" 726 726 + 727 727 + 728 728 + def test_existing_facet_merge_news_copy(ingest_env): 729 729 + env = ingest_env 730 730 + (env["root"] / "facets" / "work").mkdir(parents=True, exist_ok=True) 731 731 + 732 732 + facets = [ 733 733 + { 734 734 + "name": "work", 735 735 + "files": [ 736 736 + { 737 737 + "path": "news/20260305.md", 738 738 + "type": "news", 739 739 + "content": b"headline\n", 740 740 + } 741 741 + ], 742 742 + } 743 743 + ] 744 744 + metadata, file_map = _build_request(facets) 745 745 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 746 746 + 747 747 + target_path = env["root"] / "facets" / "work" / "news" / "20260305.md" 748 748 + assert response.status_code == 200 749 749 + assert response.get_json()["merged"] == 1 750 750 + assert target_path.read_text(encoding="utf-8") == "headline\n" 751 751 + 752 752 + 753 753 + def test_existing_facet_merge_logs(ingest_env): 754 754 + env = ingest_env 755 755 + target_path = env["root"] / "facets" / "work" / "logs" / "20260305.jsonl" 756 756 + _write_jsonl(target_path, [{"event": "existing"}]) 757 757 + 758 758 + facets = [ 759 759 + { 760 760 + "name": "work", 761 761 + "files": [ 762 762 + { 763 763 + "path": "logs/20260305.jsonl", 764 764 + "type": "logs", 765 765 + "content": _jsonl_bytes([{"event": "new"}]), 766 766 + } 767 767 + ], 768 768 + } 769 769 + ] 770 770 + metadata, file_map = _build_request(facets) 771 771 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 772 772 + 773 773 + assert response.status_code == 200 774 774 + assert response.get_json()["merged"] == 1 775 775 + assert _read_jsonl_file(target_path) == [{"event": "existing"}, {"event": "new"}] 776 776 + 777 777 + 778 778 + def test_entity_id_remapping(ingest_env): 779 779 + env = ingest_env 780 780 + facets = [ 781 781 + { 782 782 + "name": "work", 783 783 + "files": [ 784 784 + { 785 785 + "path": "entities/source_entity/entity.json", 786 786 + "type": "entity_relationship", 787 787 + "content": _json_bytes({"description": "Remapped"}), 788 788 + }, 789 789 + { 790 790 + "path": "entities/source_entity/observations.jsonl", 791 791 + "type": "entity_observations", 792 792 + "content": _jsonl_bytes([{"content": "Knows Rust", "observed_at": 1}]), 793 793 + }, 794 794 + { 795 795 + "path": "entities/20260305.jsonl", 796 796 + "type": "detected_entities", 797 797 + "content": _jsonl_bytes( 798 798 + [{"id": "source_entity", "name": "Source Entity", "type": "Person"}] 799 799 + ), 800 800 + }, 801 801 + { 802 802 + "path": "activities/20260305.jsonl", 803 803 + "type": "activity_records", 804 804 + "content": _jsonl_bytes( 805 805 + [ 806 806 + { 807 807 + "id": "coding_1", 808 808 + "activity": "coding", 809 809 + "active_entities": ["source_entity"], 810 810 + } 811 811 + ] 812 812 + ), 813 813 + }, 814 814 + ], 815 815 + } 816 816 + ] 817 817 + metadata, file_map = _build_request(facets) 818 818 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 819 819 + 820 820 + facet_root = env["root"] / "facets" / "work" 821 821 + assert response.status_code == 200 822 822 + assert response.get_json()["created"] == 4 823 823 + assert (facet_root / "entities" / "target_entity" / "entity.json").exists() 824 824 + assert (facet_root / "entities" / "target_entity" / "observations.jsonl").exists() 825 825 + assert not (facet_root / "entities" / "source_entity").exists() 826 826 + assert _read_jsonl_file(facet_root / "entities" / "20260305.jsonl")[0]["id"] == "target_entity" 827 827 + assert _read_jsonl_file(facet_root / "activities" / "20260305.jsonl")[0]["active_entities"] == [ 828 828 + "target_entity" 829 829 + ] 830 830 + 831 831 + 832 832 + def test_unmapped_entity_staging(ingest_env): 833 833 + env = ingest_env 834 834 + content = _json_bytes({"description": "Unknown"}) 835 835 + facets = [ 836 836 + { 837 837 + "name": "work", 838 838 + "files": [ 839 839 + { 840 840 + "path": "entities/unknown/entity.json", 841 841 + "type": "entity_relationship", 842 842 + "content": content, 843 843 + } 844 844 + ], 845 845 + } 846 846 + ] 847 847 + metadata, file_map = _build_request(facets) 848 848 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 849 849 + 850 850 + assert response.status_code == 200 851 851 + assert response.get_json() == { 852 852 + "created": 0, 853 853 + "merged": 0, 854 854 + "skipped": 0, 855 855 + "staged": 1, 856 856 + "errors": [], 857 857 + } 858 858 + staged = _read_staged(env["key_prefix"], "work", "entity_relationship", "entities/unknown/entity.json") 859 859 + assert staged["reason"] == "unmapped_entity" 860 860 + assert staged["source_entity_id"] == "unknown" 861 861 + assert staged["source_path"] == "entities/unknown/entity.json" 862 862 + assert "unknown" in staged["explanation"] 863 863 + 864 864 + 865 865 + def test_staged_then_retry(ingest_env): 866 866 + env = ingest_env 867 867 + facets = [ 868 868 + { 869 869 + "name": "work", 870 870 + "files": [ 871 871 + { 872 872 + "path": "entities/unknown/entity.json", 873 873 + "type": "entity_relationship", 874 874 + "content": _json_bytes({"description": "test"}), 875 875 + } 876 876 + ], 877 877 + } 878 878 + ] 879 879 + metadata, file_map = _build_request(facets) 880 880 + first = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 881 881 + 882 882 + assert first.status_code == 200 883 883 + assert first.get_json()["staged"] == 1 884 884 + 885 885 + state_path = get_state_directory(env["key_prefix"]) / "entities" / "state.json" 886 886 + entity_state = json.loads(state_path.read_text(encoding="utf-8")) 887 887 + entity_state["id_map"]["unknown"] = "unknown" 888 888 + state_path.write_text(json.dumps(entity_state), encoding="utf-8") 889 889 + 890 890 + metadata, file_map = _build_request(facets) 891 891 + second = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 892 892 + 893 893 + assert second.status_code == 200 894 894 + body = second.get_json() 895 895 + assert body["staged"] == 0 896 896 + assert body["skipped"] == 0 897 897 + assert body["created"] == 1 or body["merged"] == 1 898 898 + 899 899 + 900 900 + def test_facet_json_conflict_staging(ingest_env): 901 901 + env = ingest_env 902 902 + target_path = env["root"] / "facets" / "work" / "facet.json" 903 903 + _write_json(target_path, {"title": "Current"}) 904 904 + 905 905 + facets = [ 906 906 + { 907 907 + "name": "work", 908 908 + "files": [ 909 909 + { 910 910 + "path": "facet.json", 911 911 + "type": "facet_json", 912 912 + "content": _json_bytes({"title": "Incoming"}), 913 913 + } 914 914 + ], 915 915 + } 916 916 + ] 917 917 + metadata, file_map = _build_request(facets) 918 918 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 919 919 + 920 920 + assert response.status_code == 200 921 921 + assert response.get_json()["staged"] == 1 922 922 + staged = _read_staged(env["key_prefix"], "work", "facet_json", "facet.json") 923 923 + assert staged["reason"] == "facet_json_conflict" 924 924 + assert staged["source_content"] == {"title": "Incoming"} 925 925 + assert staged["target_content"] == {"title": "Current"} 926 926 + 927 927 + 928 928 + def test_idempotent(ingest_env): 929 929 + env = ingest_env 930 930 + facets = [ 931 931 + { 932 932 + "name": "work", 933 933 + "files": [ 934 934 + {"path": "news/20260305.md", "type": "news", "content": b"repeat\n"} 935 935 + ], 936 936 + } 937 937 + ] 938 938 + metadata, file_map = _build_request(facets) 939 939 + first = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 940 940 + 941 941 + metadata, file_map = _build_request(facets) 942 942 + second = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 943 943 + 944 944 + assert first.status_code == 200 945 945 + assert second.status_code == 200 946 946 + assert first.get_json()["created"] == 1 947 947 + assert second.get_json() == { 948 948 + "created": 0, 949 949 + "merged": 0, 950 950 + "skipped": 1, 951 951 + "staged": 0, 952 952 + "errors": [], 953 953 + } 954 954 + 955 955 + 956 956 + def test_error_isolation(ingest_env): 957 957 + env = ingest_env 958 958 + (env["root"] / "facets" / "work").mkdir(parents=True, exist_ok=True) 959 959 + facets = [ 960 960 + { 961 961 + "name": "work", 962 962 + "files": [ 963 963 + { 964 964 + "path": "entities/20260305.jsonl", 965 965 + "type": "detected_entities", 966 966 + "content": b"{bad json\n", 967 967 + }, 968 968 + {"path": "news/20260305.md", "type": "news", "content": b"good\n"}, 969 969 + ], 970 970 + } 971 971 + ] 972 972 + metadata, file_map = _build_request(facets) 973 973 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 974 974 + 975 975 + assert response.status_code == 200 976 976 + body = response.get_json() 977 977 + assert body["merged"] == 1 978 978 + assert len(body["errors"]) == 1 979 979 + assert (env["root"] / "facets" / "work" / "news" / "20260305.md").exists() 980 980 + 981 981 + 982 982 + def test_error_isolation_across_facets(ingest_env): 983 983 + env = ingest_env 984 984 + facets = [ 985 985 + { 986 986 + "name": "broken", 987 987 + "files": [ 988 988 + { 989 989 + "path": "activities/activities.jsonl", 990 990 + "type": "activity_config", 991 991 + "content": b"{bad json\n", 992 992 + } 993 993 + ], 994 994 + }, 995 995 + { 996 996 + "name": "good", 997 997 + "files": [ 998 998 + {"path": "news/20260305.md", "type": "news", "content": b"ok\n"} 999 999 + ], 1000 1000 + }, 1001 1001 + ] 1002 1002 + metadata, file_map = _build_request(facets) 1003 1003 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 1004 1004 + 1005 1005 + assert response.status_code == 200 1006 1006 + body = response.get_json() 1007 1007 + assert body["created"] == 1 1008 1008 + assert len(body["errors"]) == 1 1009 1009 + assert (env["root"] / "facets" / "good" / "news" / "20260305.md").exists() 1010 1010 + 1011 1011 + 1012 1012 + def test_stats_update(ingest_env): 1013 1013 + env = ingest_env 1014 1014 + facets = [ 1015 1015 + { 1016 1016 + "name": "alpha", 1017 1017 + "files": [ 1018 1018 + {"path": "news/20260305.md", "type": "news", "content": b"alpha\n"}, 1019 1019 + { 1020 1020 + "path": "logs/20260305.jsonl", 1021 1021 + "type": "logs", 1022 1022 + "content": _jsonl_bytes([{"event": "alpha"}]), 1023 1023 + }, 1024 1024 + ], 1025 1025 + }, 1026 1026 + { 1027 1027 + "name": "beta", 1028 1028 + "files": [ 1029 1029 + {"path": "news/20260305.md", "type": "news", "content": b"beta\n"} 1030 1030 + ], 1031 1031 + }, 1032 1032 + { 1033 1033 + "name": "gamma", 1034 1034 + "files": [ 1035 1035 + { 1036 1036 + "path": "entities/unknown/entity.json", 1037 1037 + "type": "entity_relationship", 1038 1038 + "content": _json_bytes({"description": "unknown"}), 1039 1039 + } 1040 1040 + ], 1041 1041 + }, 1042 1042 + ] 1043 1043 + metadata, file_map = _build_request(facets) 1044 1044 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 1045 1045 + source = load_journal_source(env["key"]) 1046 1046 + 1047 1047 + assert response.status_code == 200 1048 1048 + assert source["stats"]["facets_received"] == 2 1049 1049 + 1050 1050 + 1051 1051 + def test_state_manifest(ingest_env): 1052 1052 + env = ingest_env 1053 1053 + news = b"news\n" 1054 1054 + logs = _jsonl_bytes([{"event": "log"}]) 1055 1055 + facets = [ 1056 1056 + { 1057 1057 + "name": "work", 1058 1058 + "files": [ 1059 1059 + {"path": "news/20260305.md", "type": "news", "content": news}, 1060 1060 + {"path": "logs/20260305.jsonl", "type": "logs", "content": logs}, 1061 1061 + ], 1062 1062 + } 1063 1063 + ] 1064 1064 + metadata, file_map = _build_request(facets) 1065 1065 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 1066 1066 + 1067 1067 + assert response.status_code == 200 1068 1068 + assert _read_state(env["key_prefix"]) == { 1069 1069 + "received": { 1070 1070 + "work/news/20260305.md": _hash_bytes(news), 1071 1071 + "work/logs/20260305.jsonl": _hash_bytes(logs), 1072 1072 + } 1073 1073 + } 1074 1074 + 1075 1075 + 1076 1076 + def test_decision_log(ingest_env): 1077 1077 + env = ingest_env 1078 1078 + facets = [ 1079 1079 + { 1080 1080 + "name": "work", 1081 1081 + "files": [ 1082 1082 + {"path": "news/20260305.md", "type": "news", "content": b"headline\n"} 1083 1083 + ], 1084 1084 + } 1085 1085 + ] 1086 1086 + metadata, file_map = _build_request(facets) 1087 1087 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 1088 1088 + 1089 1089 + assert response.status_code == 200 1090 1090 + entries = _read_log(env["key_prefix"]) 1091 1091 + assert len(entries) == 1 1092 1092 + entry = entries[0] 1093 1093 + assert "ts" in entry 1094 1094 + assert entry["action"] == "facet_file_created" 1095 1095 + assert entry["item_type"] == "news" 1096 1096 + assert entry["item_id"] == "work/news/20260305.md" 1097 1097 + assert entry["facet"] == "work" 1098 1098 + assert entry["reason"] == "new_facet" 1099 1099 + 1100 1100 + 1101 1101 + def test_error_logged_to_decision_log(ingest_env): 1102 1102 + env = ingest_env 1103 1103 + facets = [ 1104 1104 + { 1105 1105 + "name": "work", 1106 1106 + "files": [ 1107 1107 + { 1108 1108 + "path": "todos/20260305.jsonl", 1109 1109 + "type": "todos", 1110 1110 + "content": b"{bad\n", 1111 1111 + } 1112 1112 + ], 1113 1113 + } 1114 1114 + ] 1115 1115 + metadata, file_map = _build_request(facets) 1116 1116 + response = _post_facets(env["client"], env["key"], env["key_prefix"], metadata, file_map) 1117 1117 + 1118 1118 + assert response.status_code == 200 1119 1119 + assert len(response.get_json()["errors"]) == 1 1120 1120 + 1121 1121 + entries = _read_log(env["key_prefix"]) 1122 1122 + assert len(entries) == 1 1123 1123 + assert entries[0]["action"] == "facet_file_error" 1124 1124 + assert entries[0]["facet"] == "work"