+1 -1

observe/sense.py

reviewed

··· 319 319 320 320 try: 321 321 managed = RunnerManagedProcess.spawn( 322 322 - cmd, ref=ref, callosum=self.callosum, env=env 322 322 + cmd, ref=ref, callosum=self.callosum, env=env, day=day 323 323 ) 324 324 except RuntimeError as exc: 325 325 logger.error(str(exc))

+4 -2

tests/test_activities.py

reviewed

··· 1362 1362 "work", 1363 1363 "--day", 1364 1364 "20260209", 1365 1365 - ] 1365 1365 + ], 1366 1366 + day="20260209", 1366 1367 ) 1367 1368 1368 1369 def test_ignores_wrong_tract(self): ··· 1699 1700 "--segment", 1700 1701 "100000_300", 1701 1702 "--flush", 1702 1702 - ] 1703 1703 + ], 1704 1704 + day="20260209", 1703 1705 ) 1704 1706 assert _flush_state["flushed"] is True 1705 1707

+44

tests/test_runner.py

reviewed

··· 280 280 journal_symlink = journal_path / "health" / "echo.log" 281 281 assert journal_symlink.is_symlink() 282 282 assert journal_symlink.resolve() == log_path.resolve() 283 283 + 284 284 + 285 285 + def test_process_day_override(journal_path, mock_callosum): 286 286 + """Test that day parameter overrides log directory placement.""" 287 287 + target_day = "20240101" 288 288 + managed = ManagedProcess.spawn(["echo", "day test"], day=target_day) 289 289 + ref = managed.ref 290 290 + managed.wait() 291 291 + managed.cleanup() 292 292 + 293 293 + # Log should be in target day, not today 294 294 + log_path = journal_path / target_day / "health" / f"{ref}_echo.log" 295 295 + assert log_path.exists() 296 296 + content = log_path.read_text() 297 297 + assert "day test" in content 298 298 + 299 299 + # Today's health directory should NOT have this log 300 300 + from datetime import datetime 301 301 + 302 302 + today = datetime.now().strftime("%Y%m%d") 303 303 + if today != target_day: 304 304 + today_log = journal_path / today / "health" / f"{ref}_echo.log" 305 305 + assert not today_log.exists() 306 306 + 307 307 + # Day-level symlink in target day 308 308 + day_symlink = journal_path / target_day / "health" / "echo.log" 309 309 + assert day_symlink.is_symlink() 310 310 + assert day_symlink.resolve() == log_path.resolve() 311 311 + 312 312 + # Journal-level symlink points to target day 313 313 + journal_symlink = journal_path / "health" / "echo.log" 314 314 + assert journal_symlink.is_symlink() 315 315 + assert journal_symlink.resolve() == log_path.resolve() 316 316 + 317 317 + 318 318 + def test_run_task_day_override(journal_path, mock_callosum): 319 319 + """Test that run_task passes day through to log placement.""" 320 320 + target_day = "20240201" 321 321 + success, exit_code, log_path = run_task(["echo", "task day test"], day=target_day) 322 322 + 323 323 + assert success 324 324 + assert exit_code == 0 325 325 + assert target_day in str(log_path) 326 326 + assert log_path.exists()

+2 -2

think/dream.py

reviewed

··· 80 80 cmd_name = cmd_name.replace("-", "_") 81 81 82 82 try: 83 83 - success, exit_code, _log_path = run_task(cmd) 83 83 + success, exit_code, _log_path = run_task(cmd, day=day) 84 84 if not success: 85 85 logging.error( 86 86 "Command failed with exit code %s: %s", exit_code, " ".join(cmd) ··· 127 127 listener.start(callback=on_message) 128 128 129 129 try: 130 130 - _callosum.emit("supervisor", "request", cmd=cmd, ref=ref) 130 130 + _callosum.emit("supervisor", "request", cmd=cmd, ref=ref, day=day) 131 131 132 132 if not result_event.wait(timeout=timeout): 133 133 logging.error(f"Timeout waiting for {cmd_name} to complete (ref={ref})")

+26 -15

think/runner.py

reviewed

··· 88 88 class DailyLogWriter: 89 89 """Thread-safe log writer that automatically rolls over at midnight. 90 90 91 91 + When ``day`` is provided, the writer is pinned to that day directory 92 92 + and midnight rollover is disabled (batch processing of historical days). 93 93 + 91 94 Writes to: {JOURNAL_PATH}/{YYYYMMDD}/health/{ref}_{name}.log 92 95 93 96 Creates and maintains symlinks: ··· 97 100 When the day changes, automatically closes old file, opens new file, and updates symlinks. 98 101 """ 99 102 100 100 - def __init__(self, ref: str, name: str): 103 103 + def __init__(self, ref: str, name: str, day: str | None = None): 101 104 self._ref = ref 102 105 self._name = name 106 106 + self._pinned = day is not None 103 107 self._lock = threading.Lock() 104 104 - self._current_day = _current_day() 108 108 + self._current_day = day or _current_day() 105 109 self._fh = self._open_log() 106 110 self._update_symlinks() 107 111 ··· 130 134 def write(self, message: str) -> None: 131 135 """Write message to log, handling day rollover.""" 132 136 with self._lock: 133 133 - # Check for day change 134 134 - day_now = _current_day() 135 135 - if day_now != self._current_day: 136 136 - # Close old log 137 137 - if not self._fh.closed: 138 138 - self._fh.close() 139 139 - # Open new log for new day 140 140 - self._current_day = day_now 141 141 - self._fh = self._open_log() 142 142 - # Update symlinks to point to new day's file 143 143 - self._update_symlinks() 137 137 + if not self._pinned: 138 138 + # Check for day change 139 139 + day_now = _current_day() 140 140 + if day_now != self._current_day: 141 141 + # Close old log 142 142 + if not self._fh.closed: 143 143 + self._fh.close() 144 144 + # Open new log for new day 145 145 + self._current_day = day_now 146 146 + self._fh = self._open_log() 147 147 + # Update symlinks to point to new day's file 148 148 + self._update_symlinks() 144 149 145 150 # Write and flush 146 151 self._fh.write(message) ··· 199 204 env: dict | None = None, 200 205 ref: str | None = None, 201 206 callosum: CallosumConnection | None = None, 207 207 + day: str | None = None, 202 208 ) -> "ManagedProcess": 203 209 """Spawn process with automatic output logging to daily health directory. 204 210 ··· 207 213 env: Optional environment variables (inherits parent env if not provided) 208 214 ref: Optional correlation ID (auto-generated if not provided) 209 215 callosum: Optional shared CallosumConnection (creates new one if not provided) 216 216 + day: Optional day override (YYYYMMDD). When provided, logs are placed 217 217 + in that day's health directory instead of today's. 210 218 211 219 Returns: 212 220 ManagedProcess instance ··· 243 251 callosum = CallosumConnection() 244 252 callosum.start() 245 253 246 246 - log_writer = DailyLogWriter(ref, name) 254 254 + log_writer = DailyLogWriter(ref, name, day=day) 247 255 248 256 logger.info(f"Starting {name}: {' '.join(cmd)}") 249 257 ··· 429 437 env: dict | None = None, 430 438 ref: str | None = None, 431 439 callosum: CallosumConnection | None = None, 440 440 + day: str | None = None, 432 441 ) -> tuple[bool, int, Path]: 433 442 """Run a task to completion with automatic logging (blocking). 434 443 ··· 442 451 env: Optional environment variables 443 452 ref: Optional correlation ID (auto-generated if not provided) 444 453 callosum: Optional shared CallosumConnection (creates new one if not provided) 454 454 + day: Optional day override (YYYYMMDD). When provided, logs are placed 455 455 + in that day's health directory instead of today's. 445 456 446 457 Returns: 447 458 (success, exit_code, log_path) tuple where success = (exit_code == 0) ··· 461 472 ) 462 473 # Logs to: {JOURNAL}/{YYYYMMDD}/health/1730476800000_indexer.log 463 474 """ 464 464 - managed = ManagedProcess.spawn(cmd, env=env, ref=ref, callosum=callosum) 475 475 + managed = ManagedProcess.spawn(cmd, env=env, ref=ref, callosum=callosum, day=day) 465 476 log_path = managed.log_writer.path 466 477 try: 467 478 exit_code = managed.wait(timeout=timeout)

+16 -7

think/supervisor.py

reviewed

··· 168 168 self, 169 169 cmd: list[str], 170 170 ref: str | None = None, 171 171 + day: str | None = None, 171 172 ) -> str | None: 172 173 """Submit a task for execution. 173 174 ··· 177 178 Args: 178 179 cmd: Command to execute 179 180 ref: Optional caller-provided ref for tracking 181 181 + day: Optional day override (YYYYMMDD) for log placement 180 182 181 183 Returns: 182 184 ref if task was started/queued, None if already tracked (no change) ··· 204 206 logging.debug(f"Ref already tracked for queued task: {ref}") 205 207 return None 206 208 else: 207 207 - queue.append({"refs": [ref], "cmd": cmd}) 209 209 + queue.append({"refs": [ref], "cmd": cmd, "day": day}) 208 210 logging.info( 209 211 f"Queued task {cmd_name}: {' '.join(cmd)} ref={ref} " 210 212 f"(queue: {len(queue)})" ··· 224 226 if should_start: 225 227 threading.Thread( 226 228 target=self._run_task, 227 227 - args=([ref], cmd, cmd_name), 229 229 + args=([ref], cmd, cmd_name, day), 228 230 daemon=True, 229 231 ).start() 230 232 return ref ··· 236 238 refs: list[str], 237 239 cmd: list[str], 238 240 cmd_name: str, 241 241 + day: str | None = None, 239 242 ) -> None: 240 243 """Execute a task and handle completion. 241 244 ··· 243 246 refs: List of refs to notify on completion 244 247 cmd: Command to execute 245 248 cmd_name: Command name for queue management 249 249 + day: Optional day override (YYYYMMDD) for log placement 246 250 """ 247 251 callosum = CallosumConnection() 248 252 managed = None ··· 254 258 logging.info(f"Starting task {primary_ref}: {' '.join(cmd)}") 255 259 256 260 managed = RunnerManagedProcess.spawn( 257 257 - cmd, ref=primary_ref, callosum=callosum 261 261 + cmd, ref=primary_ref, callosum=callosum, day=day 258 262 ) 259 263 self._active[primary_ref] = managed 260 264 ··· 305 309 """Process next queued task after completion.""" 306 310 next_cmd = None 307 311 refs = None 312 312 + day = None 308 313 309 314 with self._lock: 310 315 queue = self._queues.get(cmd_name, []) ··· 312 317 entry = queue.pop(0) 313 318 refs = entry["refs"] 314 319 next_cmd = entry["cmd"] 320 320 + day = entry.get("day") 315 321 self._running[cmd_name] = refs[0] 316 322 logging.info( 317 323 f"Dequeued task {cmd_name}: {' '.join(next_cmd)} refs={refs} " ··· 325 331 if next_cmd: 326 332 threading.Thread( 327 333 target=self._run_task, 328 328 - args=(refs, next_cmd, cmd_name, None), 334 334 + args=(refs, next_cmd, cmd_name, day), 329 335 daemon=True, 330 336 ).start() 331 337 ··· 644 650 return 645 651 646 652 ref = message.get("ref") 653 653 + day = message.get("day") 647 654 if _task_queue: 648 648 - _task_queue.submit(cmd, ref) 655 655 + _task_queue.submit(cmd, ref, day=day) 649 656 650 657 651 658 def _handle_supervisor_request(message: dict) -> None: ··· 1016 1023 success, exit_code, log_path = run_task( 1017 1024 ["sol", "dream", "-v", "--day", day, "--refresh"], 1018 1025 callosum=_supervisor_callosum, 1026 1026 + day=day, 1019 1027 ) 1020 1028 1021 1029 # Update state on completion ··· 1132 1140 success, exit_code, log_path = run_task( 1133 1141 cmd, 1134 1142 callosum=_supervisor_callosum, 1143 1143 + day=day, 1135 1144 ) 1136 1145 1137 1146 if success: ··· 1178 1187 if stream: 1179 1188 cmd.extend(["--stream", stream]) 1180 1189 if _task_queue: 1181 1181 - _task_queue.submit(cmd) 1190 1190 + _task_queue.submit(cmd, day=day) 1182 1191 logging.info(f"Queued segment flush: {day}/{segment}") 1183 1192 else: 1184 1193 logging.warning( ··· 1259 1268 cmd = ["sol", "dream", "--activity", record_id, "--facet", facet, "--day", day] 1260 1269 1261 1270 if _task_queue: 1262 1262 - _task_queue.submit(cmd) 1271 1271 + _task_queue.submit(cmd, day=day) 1263 1272 logging.info(f"Queued activity dream: {record_id} for #{facet}") 1264 1273 else: 1265 1274 logging.warning("No task queue available for activity dream: %s", record_id)