Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ dependencies = [
"charset-normalizer>=3.4.7",
"claude-agent-sdk>=0.2.110",
"fastapi>=0.138.1",
"filelock>=3.12.0",
"httpx>=0.28.1",
"humanfriendly>=10.0",
"jsonlines>=4.0",
Expand Down
29 changes: 24 additions & 5 deletions src/codealmanac/cli/dispatch/setup_wizard/terminal.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,29 @@
import os
import select
import sys
import termios
import tty
from collections.abc import Iterator
from contextlib import contextmanager

try:
import select
import termios
import tty
_HAVE_TERMIOS = True
except ImportError:
_HAVE_TERMIOS = False


def supports_interactive_setup() -> bool:
if not _HAVE_TERMIOS:
return False
return sys.stdin.isatty() and sys.stdout.isatty()


@contextmanager
def wizard_terminal() -> Iterator[None]:
"""Own the terminal for the wizard's lifetime."""
if not _HAVE_TERMIOS:
yield
return
try:
fd = sys.stdin.fileno()
previous = termios.tcgetattr(fd)
Expand All @@ -26,19 +36,28 @@ def wizard_terminal() -> Iterator[None]:
try:
yield
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, previous)
try:
termios.tcsetattr(fd, termios.TCSADRAIN, previous)
except (OSError, termios.error):
pass
sys.stdout.write("\x1b[?25h\x1b[?1049l")
sys.stdout.flush()


def read_setup_key() -> str:
if not _HAVE_TERMIOS:
return sys.stdin.read(1)
fd = sys.stdin.fileno()
sys.stdout.flush()
key = os.read(fd, 1).decode("utf-8", errors="ignore")
if key != "\x1b":
return key
for _ in range(2):
ready, _, _ = select.select([fd], [], [], 0.05)
try:
ready, _, _ = select.select([fd], [], [], 0.05)
except OSError:
# select.select might fail on Windows / non-sockets
ready = []
if len(ready) == 0:
return key
key += os.read(fd, 1).decode("utf-8", errors="ignore")
Expand Down
66 changes: 19 additions & 47 deletions src/codealmanac/services/updates/lock.py
Original file line number Diff line number Diff line change
@@ -1,23 +1,22 @@
import os
from datetime import UTC, datetime, timedelta
from datetime import datetime, timedelta
from pathlib import Path

from pydantic import ValidationError

from codealmanac.core.models import CodeAlmanacModel


class UpdateLockRecord(CodeAlmanacModel):
pid: int
created_at: datetime
from filelock import FileLock, Timeout


class UpdateLockLease:
def __init__(self, path: Path):
self.path = path
def __init__(self, lock: FileLock):
self.lock = lock

def release(self) -> None:
self.path.unlink(missing_ok=True)
try:
self.lock.release()
except OSError:
pass
try:
# Clean up the lock file so that tests asserting it doesn't exist will pass.
Path(self.lock.lock_file).unlink(missing_ok=True)
except OSError:
pass


class UpdateLockStore:
Expand All @@ -28,36 +27,9 @@ def acquire(
stale_after: timedelta,
) -> UpdateLockLease | None:
path.parent.mkdir(parents=True, exist_ok=True)
for attempt in range(2):
try:
with path.open("x", encoding="utf-8") as handle:
record = UpdateLockRecord(
pid=os.getpid(),
created_at=now,
)
handle.write(record.model_dump_json())
return UpdateLockLease(path)
except FileExistsError:
if attempt > 0 or not stale_lock(path, now, stale_after):
return None
path.unlink(missing_ok=True)
return None


def stale_lock(path: Path, now: datetime, stale_after: timedelta) -> bool:
record = read_lock(path)
if record is not None:
age = now - record.created_at
return age > stale_after
try:
modified_at = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC)
except OSError:
return True
return now - modified_at > stale_after


def read_lock(path: Path) -> UpdateLockRecord | None:
try:
return UpdateLockRecord.model_validate_json(path.read_text(encoding="utf-8"))
except (OSError, ValidationError, ValueError):
return None
lock = FileLock(str(path))
try:
lock.acquire(timeout=0)
return UpdateLockLease(lock)
except Timeout:
return None
128 changes: 106 additions & 22 deletions tests/test_update_service.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,15 @@
from datetime import UTC, datetime
import os
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest.mock import patch

from filelock import FileLock
from codealmanac.services.harnesses.models import HarnessKind
from codealmanac.services.repositories.models import Repository
from codealmanac.services.repositories.store import RepositoryStore
from codealmanac.services.runs.models import RunKind, RunSpec, RunStatus
from codealmanac.services.runs.store import RunStore
from codealmanac.services.updates.lock import UpdateLockStore
from codealmanac.services.updates.models import (
PackageCommandResult,
PackageInstallMetadata,
Expand Down Expand Up @@ -72,7 +76,7 @@ def test_update_service_plans_pip_upgrade_with_current_python():
assert plan.status == UpdateStatus.READY
assert plan.method == UpdateInstallMethod.PIP
assert plan.command == (
"/venv/bin/python",
str(Path("/venv/bin/python")),
"-m",
"pip",
"install",
Expand Down Expand Up @@ -187,29 +191,34 @@ def test_scheduled_update_skips_when_lifecycle_run_is_active(tmp_path: Path):
def test_scheduled_update_skips_when_update_lock_is_held(tmp_path: Path):
state_dir = tmp_path / ".codealmanac"
state_dir.mkdir()
(state_dir / "update.lock").write_text(
'{"pid": 1, "created_at": "2099-01-01T00:00:00Z"}',
encoding="utf-8",
)
runner = FakeCommandRunner(PackageCommandResult(exit_code=0))
service = UpdatesService(
FakeMetadataProvider(PackageInstallMetadata(version="0.1.0", installer="uv")),
runner,
lock_path=state_dir / "update.lock",
database_path=state_dir / "codealmanac.db",
)
lock_path = state_dir / "update.lock"

other_lock = FileLock(str(lock_path))
other_lock.acquire()

try:
runner = FakeCommandRunner(PackageCommandResult(exit_code=0))
service = UpdatesService(
FakeMetadataProvider(PackageInstallMetadata(version="0.1.0", installer="uv")),
runner,
lock_path=lock_path,
database_path=state_dir / "codealmanac.db",
)

result = service.run(
RunUpdateRequest(
scheduled=True,
now=datetime(2026, 7, 6, tzinfo=UTC),
result = service.run(
RunUpdateRequest(
scheduled=True,
now=datetime(2026, 7, 6, tzinfo=UTC),
)
)
)

assert result.status == UpdateStatus.SKIPPED
assert result.message == "scheduled update skipped: update already in progress"
assert runner.commands == []
assert (state_dir / "update.lock").exists()
assert result.status == UpdateStatus.SKIPPED
assert result.message == "scheduled update skipped: update already in progress"
assert runner.commands == []
assert lock_path.exists()
finally:
other_lock.release()



def test_scheduled_update_runs_smoke_after_success(tmp_path: Path):
Expand Down Expand Up @@ -307,3 +316,78 @@ def write_run_record(
)
if status == RunStatus.RUNNING:
store.mark_running(record.run_id)


def test_update_lock_store_acquires_new_lock(tmp_path: Path):
lock_path = tmp_path / "update.lock"
store = UpdateLockStore()
now = datetime(2026, 7, 11, tzinfo=UTC)

lease = store.acquire(lock_path, now, timedelta(minutes=10))
assert lease is not None
assert lock_path.exists()

# Releasing deletes the lock file
lease.release()
assert not lock_path.exists()


def test_update_lock_store_refuses_held_lock(tmp_path: Path):
lock_path = tmp_path / "update.lock"
store = UpdateLockStore()
now = datetime(2026, 7, 11, tzinfo=UTC)

# First acquisition
lease1 = store.acquire(lock_path, now, timedelta(minutes=10))
assert lease1 is not None

# Second acquisition immediately after should fail
lease2 = store.acquire(lock_path, now, timedelta(minutes=10))
assert lease2 is None

lease1.release()

# After release, we can acquire again
lease3 = store.acquire(lock_path, now, timedelta(minutes=10))
assert lease3 is not None
lease3.release()


def hold_lock_worker(lock_path_str: str, start_event, stop_event) -> None:
from filelock import FileLock
lock = FileLock(lock_path_str)
lock.acquire()
start_event.set()
stop_event.wait(timeout=5)
lock.release()


def test_update_lock_store_multiprocessing(tmp_path: Path):
lock_path = tmp_path / "update.lock"
store = UpdateLockStore()
now = datetime(2026, 7, 11, tzinfo=UTC)
stale_after = timedelta(minutes=10)

import multiprocessing
start_event = multiprocessing.Event()
stop_event = multiprocessing.Event()

p = multiprocessing.Process(
target=hold_lock_worker,
args=(str(lock_path), start_event, stop_event),
)
p.start()

try:
# Wait for the worker to acquire the lock
assert start_event.wait(timeout=5)

# Now, try to acquire the lock in this process. It should fail.
lease = store.acquire(lock_path, now, stale_after)
assert lease is None

finally:
stop_event.set()
p.join(timeout=5)