personal memory agent
Fix file descriptor leak in cortex subprocess pipe handling
Cortex was leaking 2 FDs (stdout + stderr) per spawned agent because
the subprocess pipes were never explicitly closed after iteration.
After ~542 agents over ~20 hours, this caused EMFILE errors blocking
all generator spawning during overnight dream runs.
The fix wraps pipe iteration in context managers (`with pipe:`) to
ensure proper cleanup, matching the pattern already used in runner.py.
Also adds MockPipe test helper to support context manager protocol
in tests that mock subprocess stdout.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+146
-117
+27
-4
tests/test_cortex.py
+27
-4
tests/test_cortex.py
···
12
12
from think.models import GPT_5
13
13
14
14
15
+
class MockPipe:
16
+
"""Mock for subprocess stdout/stderr that supports context manager protocol."""
17
+
18
+
def __init__(self, lines: list[str]):
19
+
self._lines = lines
20
+
self._iter = None
21
+
22
+
def __enter__(self):
23
+
self._iter = iter(self._lines)
24
+
return self
25
+
26
+
def __exit__(self, *args):
27
+
pass
28
+
29
+
def __iter__(self):
30
+
return self._iter or iter(self._lines)
31
+
32
+
def __next__(self):
33
+
if self._iter is None:
34
+
self._iter = iter(self._lines)
35
+
return next(self._iter)
36
+
37
+
15
38
@pytest.fixture
16
39
def mock_journal(tmp_path, monkeypatch):
17
40
"""Set up a temporary journal directory."""
···
654
677
}
655
678
}
656
679
657
-
# Create mock process with stdout
680
+
# Create mock process with stdout (MockPipe supports context manager protocol)
658
681
mock_process = MagicMock()
659
682
mock_stdout = [
660
683
'{"event": "start", "ts": 1000}\n',
661
684
'{"event": "finish", "ts": 2000, "result": "Test result"}\n',
662
685
]
663
-
mock_process.stdout = iter(mock_stdout)
686
+
mock_process.stdout = MockPipe(mock_stdout)
664
687
mock_process.wait.return_value = 0
665
688
666
689
agent = AgentProcess(agent_id, mock_process, active_path)
···
703
726
}
704
727
}
705
728
706
-
# Create mock process with stdout
729
+
# Create mock process with stdout (MockPipe supports context manager protocol)
707
730
mock_process = MagicMock()
708
731
mock_stdout = [
709
732
'{"event": "start", "ts": 1000}\n',
710
733
'{"event": "finish", "ts": 2000, "result": "Daily report content"}\n',
711
734
]
712
-
mock_process.stdout = iter(mock_stdout)
735
+
mock_process.stdout = MockPipe(mock_stdout)
713
736
mock_process.wait.return_value = 0
714
737
715
738
agent = AgentProcess(agent_id, mock_process, active_path)
+119
-113
think/cortex.py
+119
-113
think/cortex.py
···
540
540
return
541
541
542
542
try:
543
-
for line in agent.process.stdout:
544
-
if not line:
545
-
continue
543
+
with agent.process.stdout:
544
+
for line in agent.process.stdout:
545
+
if not line:
546
+
continue
546
547
547
-
line = line.strip()
548
-
if not line:
549
-
continue
548
+
line = line.strip()
549
+
if not line:
550
+
continue
550
551
551
-
try:
552
-
# Parse JSON event
553
-
event = json.loads(line)
552
+
try:
553
+
# Parse JSON event
554
+
event = json.loads(line)
554
555
555
-
# Ensure event has timestamp and agent_id
556
-
if "ts" not in event:
557
-
event["ts"] = int(time.time() * 1000)
558
-
if "agent_id" not in event:
559
-
event["agent_id"] = agent.agent_id
556
+
# Ensure event has timestamp and agent_id
557
+
if "ts" not in event:
558
+
event["ts"] = int(time.time() * 1000)
559
+
if "agent_id" not in event:
560
+
event["agent_id"] = agent.agent_id
560
561
561
-
# Append to JSONL file
562
-
with open(agent.log_path, "a") as f:
563
-
f.write(json.dumps(event) + "\n")
562
+
# Append to JSONL file
563
+
with open(agent.log_path, "a") as f:
564
+
f.write(json.dumps(event) + "\n")
564
565
565
-
# Broadcast event to Callosum
566
-
try:
567
-
event_copy = event.copy()
568
-
event_type = event_copy.pop("event", "unknown")
569
-
self.callosum.emit("cortex", event_type, **event_copy)
570
-
except Exception as e:
571
-
self.logger.info(f"Failed to broadcast event to Callosum: {e}")
566
+
# Broadcast event to Callosum
567
+
try:
568
+
event_copy = event.copy()
569
+
event_type = event_copy.pop("event", "unknown")
570
+
self.callosum.emit("cortex", event_type, **event_copy)
571
+
except Exception as e:
572
+
self.logger.info(
573
+
f"Failed to broadcast event to Callosum: {e}"
574
+
)
572
575
573
-
# Capture model from start event (needed for token usage logging)
574
-
if event.get("event") == "start":
575
-
model = event.get("model")
576
-
if model:
577
-
with self.lock:
578
-
if agent.agent_id in self.agent_requests:
579
-
self.agent_requests[agent.agent_id]["model"] = model
576
+
# Capture model from start event (needed for token usage logging)
577
+
if event.get("event") == "start":
578
+
model = event.get("model")
579
+
if model:
580
+
with self.lock:
581
+
if agent.agent_id in self.agent_requests:
582
+
self.agent_requests[agent.agent_id][
583
+
"model"
584
+
] = model
580
585
581
-
# Handle finish or error event
582
-
if event.get("event") in ["finish", "error"]:
583
-
# Check for output and handoff (only on finish)
584
-
if event.get("event") == "finish":
585
-
result = event.get("result", "")
586
+
# Handle finish or error event
587
+
if event.get("event") in ["finish", "error"]:
588
+
# Check for output and handoff (only on finish)
589
+
if event.get("event") == "finish":
590
+
result = event.get("result", "")
586
591
587
-
# Get original request (thread-safe access)
588
-
with self.lock:
589
-
original_request = self.agent_requests.get(
590
-
agent.agent_id
591
-
)
592
+
# Get original request (thread-safe access)
593
+
with self.lock:
594
+
original_request = self.agent_requests.get(
595
+
agent.agent_id
596
+
)
592
597
593
-
# Log token usage if available
594
-
usage_data = event.get("usage")
595
-
if usage_data and original_request:
596
-
try:
597
-
from think.models import log_token_usage
598
+
# Log token usage if available
599
+
usage_data = event.get("usage")
600
+
if usage_data and original_request:
601
+
try:
602
+
from think.models import log_token_usage
598
603
599
-
model = original_request.get("model", "unknown")
600
-
name = original_request.get("name", "unknown")
604
+
model = original_request.get("model", "unknown")
605
+
name = original_request.get("name", "unknown")
601
606
602
-
# Build context in same format as model resolution:
603
-
# agent.{app}.{name} where app="system" for system agents
604
-
if ":" in name:
605
-
app, name = name.split(":", 1)
606
-
else:
607
-
app, name = "system", name
608
-
context = f"agent.{app}.{name}"
607
+
# Build context in same format as model resolution:
608
+
# agent.{app}.{name} where app="system" for system agents
609
+
if ":" in name:
610
+
app, name = name.split(":", 1)
611
+
else:
612
+
app, name = "system", name
613
+
context = f"agent.{app}.{name}"
609
614
610
-
# Extract segment from env if set (flat merge puts env at top level)
611
-
env_config = original_request.get("env", {})
612
-
segment = (
613
-
env_config.get("SEGMENT_KEY")
614
-
if env_config
615
-
else None
616
-
)
615
+
# Extract segment from env if set (flat merge puts env at top level)
616
+
env_config = original_request.get("env", {})
617
+
segment = (
618
+
env_config.get("SEGMENT_KEY")
619
+
if env_config
620
+
else None
621
+
)
617
622
618
-
log_token_usage(
619
-
model=model,
620
-
usage=usage_data,
621
-
context=context,
622
-
segment=segment,
623
-
)
624
-
except Exception as e:
625
-
self.logger.warning(
626
-
f"Failed to log token usage for agent {agent.agent_id}: {e}"
623
+
log_token_usage(
624
+
model=model,
625
+
usage=usage_data,
626
+
context=context,
627
+
segment=segment,
628
+
)
629
+
except Exception as e:
630
+
self.logger.warning(
631
+
f"Failed to log token usage for agent {agent.agent_id}: {e}"
632
+
)
633
+
634
+
# Write output if requested
635
+
if original_request and original_request.get("output"):
636
+
self._write_output(
637
+
agent.agent_id,
638
+
result,
639
+
original_request,
627
640
)
628
641
629
-
# Write output if requested
630
-
if original_request and original_request.get("output"):
631
-
self._write_output(
632
-
agent.agent_id,
633
-
result,
634
-
original_request,
635
-
)
642
+
# Handle handoff (prefer stored config captured at startup)
643
+
handoff_config = None
644
+
with self.lock:
645
+
if agent.agent_id in self.agent_handoffs:
646
+
handoff_config = copy.deepcopy(
647
+
self.agent_handoffs.pop(agent.agent_id)
648
+
)
636
649
637
-
# Handle handoff (prefer stored config captured at startup)
638
-
handoff_config = None
639
-
with self.lock:
640
-
if agent.agent_id in self.agent_handoffs:
641
-
handoff_config = copy.deepcopy(
642
-
self.agent_handoffs.pop(agent.agent_id)
650
+
if handoff_config:
651
+
self._spawn_handoff(
652
+
agent.agent_id, result, handoff_config
643
653
)
654
+
# Break to trigger cleanup
655
+
break
644
656
645
-
if handoff_config:
646
-
self._spawn_handoff(
647
-
agent.agent_id, result, handoff_config
648
-
)
649
-
# Break to trigger cleanup
650
-
break
651
-
652
-
except json.JSONDecodeError:
653
-
# Non-JSON output becomes info event
654
-
info_event = {
655
-
"event": "info",
656
-
"ts": int(time.time() * 1000),
657
-
"message": line,
658
-
"agent_id": agent.agent_id,
659
-
}
660
-
with open(agent.log_path, "a") as f:
661
-
f.write(json.dumps(info_event) + "\n")
657
+
except json.JSONDecodeError:
658
+
# Non-JSON output becomes info event
659
+
info_event = {
660
+
"event": "info",
661
+
"ts": int(time.time() * 1000),
662
+
"message": line,
663
+
"agent_id": agent.agent_id,
664
+
}
665
+
with open(agent.log_path, "a") as f:
666
+
f.write(json.dumps(info_event) + "\n")
662
667
663
668
except Exception as e:
664
669
self.logger.error(
···
702
707
703
708
stderr_lines = []
704
709
try:
705
-
for line in agent.process.stderr:
706
-
if not line:
707
-
continue
708
-
stripped = line.strip()
709
-
if stripped:
710
-
stderr_lines.append(stripped)
711
-
# Pass through to cortex stderr with agent prefix for traceability
712
-
print(
713
-
f"[agent:{agent.agent_id}:stderr] {stripped}",
714
-
file=sys.stderr,
715
-
flush=True,
716
-
)
710
+
with agent.process.stderr:
711
+
for line in agent.process.stderr:
712
+
if not line:
713
+
continue
714
+
stripped = line.strip()
715
+
if stripped:
716
+
stderr_lines.append(stripped)
717
+
# Pass through to cortex stderr with agent prefix for traceability
718
+
print(
719
+
f"[agent:{agent.agent_id}:stderr] {stripped}",
720
+
file=sys.stderr,
721
+
flush=True,
722
+
)
717
723
718
724
except Exception as e:
719
725
self.logger.error(