"""muse#53 item 6 -- relative time syntax for symlog @{...} refs. `@{1.hour.ago}`, `@{2.days.ago}`, `@{yesterday}` resolve to the most recent symlog entry at or before that point in time, exercised at the `resolve_symlog_addr` core level (with fabricated timestamps written directly to the symlog file for determinism) and through the CLI (`muse symlog resolve`). """ from __future__ import annotations import datetime import json import os import pathlib import textwrap import pytest from tests.cli_test_helper import CliRunner from muse.core.symlog import resolve_symlog_addr, symlog_path, NULL_CONTENT_ID runner = CliRunner() def _invoke(repo: pathlib.Path, args: list[str]): saved = os.getcwd() try: os.chdir(repo) return runner.invoke(None, args) finally: os.chdir(saved) def _init_repo(repo: pathlib.Path) -> None: repo.mkdir(parents=True, exist_ok=True) _invoke(repo, ["init"]) def _write_file(repo: pathlib.Path, rel: str, content: str) -> None: p = repo / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_text(textwrap.dedent(content)) def _commit(repo: pathlib.Path, msg: str) -> str: _invoke(repo, ["code", "add", "."]) result = _invoke(repo, [ "commit", "-m", msg, "--agent-id", "claude-code", "--model-id", "claude-sonnet-4-6", "--author", "claude-code", "--json", ]) data = json.loads(result.stdout) return data["commit_id"] def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: _init_repo(tmp_path) return tmp_path def _write_raw_entry(path: pathlib.Path, old_id: str, new_id: str, commit_id: str, when: datetime.datetime, author: str, operation: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) ts = int(when.timestamp()) line = f"{old_id} {new_id} {commit_id} {author} {ts} +0000\t{operation}\n" with path.open("a", encoding="utf-8") as f: f.write(line) def _fake_id(n: int) -> str: return f"sha256:{n:064x}" class TestSymlogTimeAgoResolutionCore: def test_hour_ago_finds_entry_before_cutoff(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) addr = "src/billing.py::compute_total" log_path = symlog_path(repo, addr) now = datetime.datetime.now(datetime.timezone.utc) c1 = _fake_id(1) c2 = _fake_id(2) _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now - datetime.timedelta(hours=3), "gabriel", "symbol-created: compute_total") _write_raw_entry(log_path, c1, c2, _fake_id(102), now - datetime.timedelta(minutes=10), "gabriel", "symbol-modified: fix") resolution = resolve_symlog_addr(f"{addr}@{{1.hour.ago}}", repo) assert resolution is not None assert resolution.content_id == c1 def test_minutes_ago_finds_most_recent(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) addr = "src/billing.py::compute_total" log_path = symlog_path(repo, addr) now = datetime.datetime.now(datetime.timezone.utc) c1 = _fake_id(1) c2 = _fake_id(2) _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now - datetime.timedelta(hours=2), "gabriel", "symbol-created: compute_total") _write_raw_entry(log_path, c1, c2, _fake_id(102), now - datetime.timedelta(hours=1), "gabriel", "symbol-modified: fix") resolution = resolve_symlog_addr(f"{addr}@{{30.minute.ago}}", repo) assert resolution is not None assert resolution.content_id == c2 def test_yesterday_finds_entry_before_todays_midnight(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) addr = "src/billing.py::compute_total" log_path = symlog_path(repo, addr) now = datetime.datetime.now(datetime.timezone.utc) today_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) c1 = _fake_id(1) c2 = _fake_id(2) _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), today_midnight - datetime.timedelta(days=1), "gabriel", "symbol-created: compute_total") _write_raw_entry(log_path, c1, c2, _fake_id(102), now, "gabriel", "symbol-modified: fix") resolution = resolve_symlog_addr(f"{addr}@{{yesterday}}", repo) assert resolution is not None assert resolution.content_id == c1 def test_no_entry_old_enough_raises_index_error(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) addr = "src/billing.py::compute_total" log_path = symlog_path(repo, addr) now = datetime.datetime.now(datetime.timezone.utc) c1 = _fake_id(1) _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now, "gabriel", "symbol-created: compute_total") with pytest.raises(IndexError): resolve_symlog_addr(f"{addr}@{{1.week.ago}}", repo) def test_unrecognized_spec_raises_value_error(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) addr = "src/billing.py::compute_total" log_path = symlog_path(repo, addr) now = datetime.datetime.now(datetime.timezone.utc) c1 = _fake_id(1) _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now, "gabriel", "symbol-created: compute_total") with pytest.raises(ValueError): resolve_symlog_addr(f"{addr}@{{not.a.real.spec}}", repo) def test_plain_index_still_works(self, tmp_path: pathlib.Path) -> None: """Non-regression: the pre-existing @{N} integer form must still work after broadening the regex to accept time specs.""" repo = _make_repo(tmp_path) addr = "src/billing.py::compute_total" log_path = symlog_path(repo, addr) now = datetime.datetime.now(datetime.timezone.utc) c1 = _fake_id(1) _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now, "gabriel", "symbol-created: compute_total") resolution = resolve_symlog_addr(f"{addr}@{{0}}", repo) assert resolution is not None assert resolution.content_id == c1 class TestSymlogTimeAgoViaCli: def test_resolve_accepts_time_spec(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_file(repo, "src/billing.py", "def compute_total(items):\n return sum(items)\n") _commit(repo, "add compute_total") result = _invoke(repo, ["symlog", "resolve", "src/billing.py::compute_total@{0.second.ago}", "--json"]) assert result.exit_code == 0, result.stdout data = json.loads(result.stdout) assert data["symbol"] == "src/billing.py::compute_total"