Fix manual status updates for application URLs

CHANGELOG.md

@@ -5,6 +5,13 @@ All notable changes to ApplyPilot will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+- **Manual application status accuracy** - `apply --mark-applied` and `--mark-failed`
+  now accept canonical or application URLs and fail clearly instead of reporting success
+  when no database row was updated.
+
 ## [0.2.0] - 2026-02-17
 
 ### Added

src/applypilot/apply/launcher.py

@@ -247,29 +247,56 @@ def gen_prompt(target_url: str, min_score: int = 7,
     return prompt_file
 
 
-def mark_job(url: str, status: str, reason: str | None = None) -> None:
+def mark_job(url: str, status: str, reason: str | None = None) -> str:
     """Manually mark a job's apply status in the database.
 
     Args:
         url: Job URL to mark.
         status: Either 'applied' or 'failed'.
         reason: Failure reason (only for status='failed').
+
+    Returns:
+        The canonical job URL that was updated.
+
+    Raises:
+        LookupError: No job matches the supplied canonical or application URL.
+        ValueError: The status is invalid or an application URL matches multiple jobs.
     """
+    if status not in {"applied", "failed"}:
+        raise ValueError("status must be 'applied' or 'failed'")
+
     conn = get_connection()
+    row = conn.execute("SELECT url FROM jobs WHERE url = ?", (url,)).fetchone()
+    if row is None:
+        rows = conn.execute(
+            "SELECT url FROM jobs WHERE application_url = ? ORDER BY url LIMIT 2",
+            (url,),
+        ).fetchall()
+        if not rows:
+            raise LookupError(f"No job found for URL: {url}")
+        if len(rows) > 1:
+            raise ValueError(
+                "Application URL matches multiple jobs; use the canonical job URL instead."
+            )
+        canonical_url = rows[0]["url"]
+    else:
+        canonical_url = row["url"]
+
     now = datetime.now(timezone.utc).isoformat()
     if status == "applied":
         conn.execute("""
             UPDATE jobs SET apply_status = 'applied', applied_at = ?,
                            apply_error = NULL, agent_id = NULL
             WHERE url = ?
-        """, (now, url))
+        """, (now, canonical_url))
     else:
         conn.execute("""
             UPDATE jobs SET apply_status = 'failed', apply_error = ?,
                            apply_attempts = 99, agent_id = NULL
             WHERE url = ?
-        """, (reason or "manual", url))
+        """, (reason or "manual", canonical_url))
     conn.commit()
+    return canonical_url
 
 
 def reset_failed() -> int:

src/applypilot/cli.py

@@ -168,14 +168,25 @@ def apply(
 
     if mark_applied:
         from applypilot.apply.launcher import mark_job
-        mark_job(mark_applied, "applied")
-        console.print(f"[green]Marked as applied:[/green] {mark_applied}")
+        try:
+            canonical_url = mark_job(mark_applied, "applied")
+        except (LookupError, ValueError) as exc:
+            console.print(f"[red]Could not mark job as applied:[/red] {exc}")
+            raise typer.Exit(code=1) from exc
+        console.print(f"[green]Marked as applied:[/green] {canonical_url}")
         return
 
     if mark_failed:
         from applypilot.apply.launcher import mark_job
-        mark_job(mark_failed, "failed", reason=fail_reason)
-        console.print(f"[yellow]Marked as failed:[/yellow] {mark_failed} ({fail_reason or 'manual'})")
+        try:
+            canonical_url = mark_job(mark_failed, "failed", reason=fail_reason)
+        except (LookupError, ValueError) as exc:
+            console.print(f"[red]Could not mark job as failed:[/red] {exc}")
+            raise typer.Exit(code=1) from exc
+        console.print(
+            f"[yellow]Marked as failed:[/yellow] {canonical_url} "
+            f"({fail_reason or 'manual'})"
+        )
         return
 
     if reset_failed:

tests/test_manual_status.py

@@ -0,0 +1,114 @@
+"""Regression tests for manual application status updates."""
+
+import sqlite3
+
+import pytest
+from typer.testing import CliRunner
+
+from applypilot import cli
+from applypilot.apply import launcher
+
+
+@pytest.fixture
+def jobs_db(monkeypatch: pytest.MonkeyPatch) -> sqlite3.Connection:
+    """Provide the minimal jobs schema used by ``mark_job``."""
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        """
+        CREATE TABLE jobs (
+            url TEXT PRIMARY KEY,
+            application_url TEXT,
+            applied_at TEXT,
+            apply_status TEXT,
+            apply_error TEXT,
+            apply_attempts INTEGER DEFAULT 0,
+            agent_id TEXT
+        )
+        """
+    )
+    monkeypatch.setattr(launcher, "get_connection", lambda: conn)
+    yield conn
+    conn.close()
+
+
+def _insert_job(conn: sqlite3.Connection, url: str, application_url: str) -> None:
+    conn.execute(
+        "INSERT INTO jobs (url, application_url) VALUES (?, ?)",
+        (url, application_url),
+    )
+    conn.commit()
+
+
+def test_mark_job_accepts_canonical_url(jobs_db: sqlite3.Connection) -> None:
+    canonical_url = "https://board.example/jobs/123"
+    _insert_job(jobs_db, canonical_url, "https://ats.example/apply/123")
+
+    updated_url = launcher.mark_job(canonical_url, "applied")
+
+    row = jobs_db.execute(
+        "SELECT apply_status, applied_at FROM jobs WHERE url = ?", (canonical_url,)
+    ).fetchone()
+    assert updated_url == canonical_url
+    assert row["apply_status"] == "applied"
+    assert row["applied_at"] is not None
+
+
+def test_mark_job_resolves_application_url(jobs_db: sqlite3.Connection) -> None:
+    canonical_url = "https://board.example/jobs/123"
+    application_url = "https://ats.example/apply/123"
+    _insert_job(jobs_db, canonical_url, application_url)
+
+    updated_url = launcher.mark_job(application_url, "failed", reason="manual email")
+
+    row = jobs_db.execute(
+        "SELECT apply_status, apply_error, apply_attempts FROM jobs WHERE url = ?",
+        (canonical_url,),
+    ).fetchone()
+    assert updated_url == canonical_url
+    assert row["apply_status"] == "failed"
+    assert row["apply_error"] == "manual email"
+    assert row["apply_attempts"] == 99
+
+
+def test_mark_job_rejects_unknown_url(jobs_db: sqlite3.Connection) -> None:
+    with pytest.raises(LookupError, match="No job found"):
+        launcher.mark_job("https://ats.example/apply/missing", "applied")
+
+
+def test_mark_job_rejects_ambiguous_application_url(jobs_db: sqlite3.Connection) -> None:
+    application_url = "https://ats.example/apply/shared"
+    _insert_job(jobs_db, "https://board.example/jobs/123", application_url)
+    _insert_job(jobs_db, "https://board.example/jobs/456", application_url)
+
+    with pytest.raises(ValueError, match="matches multiple jobs"):
+        launcher.mark_job(application_url, "applied")
+
+    statuses = jobs_db.execute("SELECT apply_status FROM jobs ORDER BY url").fetchall()
+    assert [row["apply_status"] for row in statuses] == [None, None]
+
+
+def test_mark_job_rejects_invalid_status(jobs_db: sqlite3.Connection) -> None:
+    with pytest.raises(ValueError, match="status must be"):
+        launcher.mark_job("https://board.example/jobs/123", "previewed")
+
+
+def test_mark_applied_cli_exits_when_no_job_matches(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The CLI must not print a success message after a zero-row update."""
+    monkeypatch.setattr(cli, "_bootstrap", lambda: None)
+
+    def missing_job(*_args: object, **_kwargs: object) -> str:
+        raise LookupError("No job found for URL: https://example.invalid/missing")
+
+    monkeypatch.setattr(launcher, "mark_job", missing_job)
+
+    result = CliRunner().invoke(
+        cli.app,
+        ["apply", "--mark-applied", "https://example.invalid/missing"],
+    )
+
+    assert result.exit_code == 1
+    assert "Could not mark job as applied" in result.output
+    assert "Marked as applied" not in result.output