diff --git a/.gitignore b/.gitignore index 835589f15..169835071 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ # User data (NEVER commit) +MY_PROFILE.md profile.json resume.txt resume.pdf @@ -19,6 +20,9 @@ logs/ .mcp*.json # Python +venv/ +.venv/ +env/ __pycache__/ *.py[cod] *$py.class diff --git a/src/applypilot/apply/launcher.py b/src/applypilot/apply/launcher.py index 341a11a36..d3f6d4b08 100644 --- a/src/applypilot/apply/launcher.py +++ b/src/applypilot/apply/launcher.py @@ -111,7 +111,7 @@ def acquire_job(target_url: str | None = None, min_score: int = 7, FROM jobs WHERE (url = ? OR application_url = ? OR application_url LIKE ? OR url LIKE ?) AND tailored_resume_path IS NOT NULL - AND apply_status != 'in_progress' + AND (apply_status IS NULL OR apply_status != 'in_progress') LIMIT 1 """, (target_url, target_url, like, like)).fetchone() else: @@ -327,6 +327,9 @@ def run_job(job: dict, port: int, worker_id: int = 0, "--model", model, "-p", "--mcp-config", str(mcp_config_path), + # WARNING: Using bypassPermissions poses a significant security risk as Claude Code + # will have unprompted access to the system. For production use, Docker isolation + # is highly recommended. "--permission-mode", "bypassPermissions", "--no-session-persistence", "--disallowedTools", ( @@ -465,7 +468,7 @@ def run_job(job: dict, port: int, worker_id: int = 0, def _clean_reason(s: str) -> str: return re.sub(r'[*`"]+$', '', s).strip() - for result_status in ["APPLIED", "EXPIRED", "CAPTCHA", "LOGIN_ISSUE"]: + for result_status in ["DRY_RUN_COMPLETE", "APPLIED", "EXPIRED", "CAPTCHA", "LOGIN_ISSUE"]: if f"RESULT:{result_status}" in output: add_event(f"[W{worker_id}] {result_status} ({elapsed}s): {job['title'][:30]}") update_state(worker_id, status=result_status.lower(), @@ -604,7 +607,15 @@ def worker_loop(worker_id: int = 0, limit: int = 1, result, duration_ms = run_job(job, port=port, worker_id=worker_id, model=model, dry_run=dry_run) - if result == "skipped": + # A dry run is observational: never write APPLIED/FAILED state, + # even if an agent returns an incorrect result token. Releasing the + # acquisition lock leaves the real queue exactly as it was. + if dry_run: + release_lock(job["url"]) + add_event(f"[W{worker_id}] DRY RUN complete: {job['title'][:30]}") + update_state(worker_id, status="dry_run_complete", + last_action="not submitted") + elif result == "skipped": release_lock(job["url"]) add_event(f"[W{worker_id}] Skipped: {job['title'][:30]}") continue @@ -781,10 +792,16 @@ def _refresh(): live.update(render_full()) totals = get_totals() - console.print( - f"\n[bold]Done: {total_applied} applied, {total_failed} failed " - f"(${totals['cost']:.3f})[/bold]" - ) + if dry_run: + console.print( + f"\n[bold]Dry run done: 0 submitted, {total_failed} execution errors " + f"(${totals['cost']:.3f})[/bold]" + ) + else: + console.print( + f"\n[bold]Done: {total_applied} applied, {total_failed} failed " + f"(${totals['cost']:.3f})[/bold]" + ) console.print(f"Logs: {config.LOG_DIR}") except KeyboardInterrupt: diff --git a/src/applypilot/apply/prompt.py b/src/applypilot/apply/prompt.py index 37c3790a1..5ac951082 100644 --- a/src/applypilot/apply/prompt.py +++ b/src/applypilot/apply/prompt.py @@ -507,13 +507,20 @@ def build_prompt(job: dict, tailored_resume: str, last_name = full_name.split()[-1] if " " in full_name else "" display_name = f"{preferred_name} {last_name}".strip() - # Dry-run: override submit instruction + # Dry-run: make every externally-visible action inert, not only the final + # submit button. Email-only applications are submissions too. if dry_run: - submit_instruction = "IMPORTANT: Do NOT click the final Submit/Apply button. Review the form, verify all fields, then output RESULT:APPLIED with a note that this was a dry run." + mission = "Inspect and fill the application for verification only. Never submit it or send any email." + email_instruction = "Do NOT send email. Record that this application requires manual email submission, then output RESULT:DRY_RUN_COMPLETE." + submit_instruction = "IMPORTANT: Do NOT click the final Submit/Apply button. Stop on the final review page, verify all fields, then output RESULT:DRY_RUN_COMPLETE." + after_submit_instruction = "Do not perform any post-submit steps because no submission is allowed in dry-run mode." else: + mission = "Submit a complete, accurate application." + email_instruction = f'send_email with subject "Application for {job["title"]} -- {display_name}", body = 2-3 sentence pitch + contact info, attach resume PDF: ["{pdf_path}"]\n - Output RESULT:APPLIED. Done.' submit_instruction = "BEFORE clicking Submit/Apply, take a snapshot and review EVERY field on the page. Verify all data matches the APPLICANT PROFILE and TAILORED RESUME -- name, email, phone, location, work auth, resume uploaded, cover letter if applicable. If anything is wrong or missing, fix it FIRST. Only click Submit after confirming everything is correct." + after_submit_instruction = "After submit: browser_snapshot. Run CAPTCHA DETECT -- submit buttons often trigger invisible CAPTCHAs. If found, solve it (the form will auto-submit once the token clears, or you may need to click Submit again). Then check for new tabs (browser_tabs action: \"list\"). Switch to newest, close old. Snapshot to confirm submission. Look for \"thank you\" or \"application received\"." - prompt = f"""You are an autonomous job application agent. Your ONE mission: get this candidate an interview. You have all the information and tools. Think strategically. Act decisively. Submit the application. + prompt = f"""You are an autonomous job application agent. Your ONE mission: get this candidate an interview. You have all the information and tools. Think strategically and act accurately. {mission} == JOB == URL: {job.get('application_url') or job['url']} @@ -535,9 +542,9 @@ def build_prompt(job: dict, tailored_resume: str, {profile_summary} == YOUR MISSION == -Submit a complete, accurate application. Use the profile and resume as source data -- adapt to fit each form's format. +{mission} Use the profile and resume as source data -- adapt to fit each form's format. -If something unexpected happens and these instructions don't cover it, figure it out yourself. You are autonomous. Navigate pages, read content, try buttons, explore the site. The goal is always the same: submit the application. Do whatever it takes to reach that goal. +If something unexpected happens and these instructions don't cover it, navigate carefully and preserve the dry-run/submission boundary above. {hard_rules} @@ -562,8 +569,7 @@ def build_prompt(job: dict, tailored_resume: str, 2. browser_snapshot to read the page. Then run CAPTCHA DETECT (see CAPTCHA section). If a CAPTCHA is found, solve it before continuing. 3. LOCATION CHECK. Read the page for location info. If not eligible, output RESULT and stop. 4. Find and click the Apply button. If email-only (page says "email resume to X"): - - send_email with subject "Application for {job['title']} -- {display_name}", body = 2-3 sentence pitch + contact info, attach resume PDF: ["{pdf_path}"] - - Output RESULT:APPLIED. Done. + - {email_instruction} After clicking Apply: browser_snapshot. Run CAPTCHA DETECT -- many sites trigger CAPTCHAs right after the Apply click. If found, solve before continuing. 5. Login wall? 5a. FIRST: check the URL. If you landed on {', '.join(blocked_sso)}, or any SSO/OAuth page -> STOP. Output RESULT:FAILED:sso_required. Do NOT try to sign in to Google/Microsoft/SSO. @@ -581,10 +587,11 @@ def build_prompt(job: dict, tailored_resume: str, - Compare every other field to the APPLICANT PROFILE. Fix mismatches. Fill empty fields. 9. Answer screening questions using the rules above. 10. {submit_instruction} -11. After submit: browser_snapshot. Run CAPTCHA DETECT -- submit buttons often trigger invisible CAPTCHAs. If found, solve it (the form will auto-submit once the token clears, or you may need to click Submit again). Then check for new tabs (browser_tabs action: "list"). Switch to newest, close old. Snapshot to confirm submission. Look for "thank you" or "application received". +11. {after_submit_instruction} 12. Output your result. == RESULT CODES (output EXACTLY one) == +RESULT:DRY_RUN_COMPLETE -- form verified but deliberately not submitted (dry-run only) RESULT:APPLIED -- submitted successfully RESULT:EXPIRED -- job closed or no longer accepting applications RESULT:CAPTCHA -- blocked by unsolvable captcha diff --git a/src/applypilot/cli.py b/src/applypilot/cli.py index 6c8be9128..f9e5177f9 100644 --- a/src/applypilot/cli.py +++ b/src/applypilot/cli.py @@ -161,7 +161,7 @@ def apply( """Launch auto-apply to submit job applications.""" _bootstrap() - from applypilot.config import check_tier, PROFILE_PATH as _profile_path + from applypilot.config import check_tier, PROFILE_PATH as _profile_path, validate_profile_for_application from applypilot.database import get_connection # --- Utility modes (no Chrome/Claude needed) --- @@ -197,6 +197,12 @@ def apply( ) raise typer.Exit(code=1) + try: + validate_profile_for_application() + except (ValueError, KeyError) as error: + console.print(f"[red]Unsafe or incomplete profile.[/red] {error}\nRun [bold]applypilot init[/bold] and review every field before applying.") + raise typer.Exit(code=1) + # Check 3: Tailored resumes exist (skip for --gen with --url) if not (gen and url): conn = get_connection() @@ -223,6 +229,9 @@ def apply( mcp_path = _profile_path.parent / ".mcp-apply-0.json" console.print(f"[green]Wrote prompt to:[/green] {prompt_file}") console.print(f"\n[bold]Run manually:[/bold]") + # WARNING: Using bypassPermissions poses a significant security risk as Claude Code + # will have unprompted access to the system. For production use, Docker isolation + # is highly recommended. console.print( f" claude --model {model} -p " f"--mcp-config {mcp_path} " @@ -338,7 +347,7 @@ def doctor() -> None: import shutil from applypilot.config import ( load_env, PROFILE_PATH, RESUME_PATH, RESUME_PDF_PATH, - SEARCH_CONFIG_PATH, ENV_PATH, get_chrome_path, + SEARCH_CONFIG_PATH, ENV_PATH, get_chrome_path, profile_safety_reasons, ) load_env() @@ -352,7 +361,14 @@ def doctor() -> None: # --- Tier 1 checks --- # Profile if PROFILE_PATH.exists(): - results.append(("profile.json", ok_mark, str(PROFILE_PATH))) + try: + profile_reasons = profile_safety_reasons() + except Exception as error: + profile_reasons = [str(error)] + if profile_reasons: + results.append(("profile.json", "[red]UNSAFE[/red]", "; ".join(profile_reasons))) + else: + results.append(("profile.json", ok_mark, str(PROFILE_PATH))) else: results.append(("profile.json", fail_mark, "Run 'applypilot init' to create")) diff --git a/src/applypilot/config.py b/src/applypilot/config.py index 8c3978073..69f1882da 100644 --- a/src/applypilot/config.py +++ b/src/applypilot/config.py @@ -101,6 +101,36 @@ def load_profile() -> dict: return json.loads(PROFILE_PATH.read_text(encoding="utf-8")) +def profile_safety_reasons(profile: dict | None = None) -> list[str]: + """Return reasons a profile must not be used for an external application.""" + data = profile if profile is not None else load_profile() + personal = data.get("personal") if isinstance(data, dict) else None + reasons: list[str] = [] + if not isinstance(personal, dict): + reasons.append("legacy/incomplete profile schema; rerun applypilot init") + personal = data if isinstance(data, dict) else {} + name = str(personal.get("full_name") or personal.get("name") or "").strip().lower() + email = str(personal.get("email") or "").strip().lower() + if not name or not email: + reasons.append("full legal name and email are required") + if any(token in name for token in ("sample", "test candidate", "firstname lastname", "your_legal_name")): + reasons.append("name contains a sample/test placeholder") + if email.startswith("youremail@") or email.endswith("@example.com") or email.endswith("@example.invalid"): + reasons.append("email contains a sample/test placeholder") + for section in ("work_authorization", "compensation", "experience", "resume_facts"): + if not isinstance(data.get(section), dict): + reasons.append(f"required profile section missing: {section}") + return list(dict.fromkeys(reasons)) + + +def validate_profile_for_application(profile: dict | None = None) -> dict: + data = profile if profile is not None else load_profile() + reasons = profile_safety_reasons(data) + if reasons: + raise ValueError("Profile is unsafe for application: " + "; ".join(reasons)) + return data + + def load_search_config() -> dict: """Load search configuration from ~/.applypilot/searches.yaml.""" import yaml @@ -175,9 +205,9 @@ def load_env(): """Load environment variables from ~/.applypilot/.env if it exists.""" from dotenv import load_dotenv if ENV_PATH.exists(): - load_dotenv(ENV_PATH) + load_dotenv(ENV_PATH, override=True) # Also try CWD .env as fallback - load_dotenv() + load_dotenv(override=True) # --------------------------------------------------------------------------- diff --git a/src/applypilot/database.py b/src/applypilot/database.py index a1779c02a..cf9202030 100644 --- a/src/applypilot/database.py +++ b/src/applypilot/database.py @@ -406,7 +406,7 @@ def get_jobs_by_stage(conn: sqlite3.Connection | None = None, elif "?" in where: params.append(7) # default min_score - if min_score is not None and "fit_score" not in where and stage in ("scored", "tailored", "applied"): + if min_score is not None and stage in ("scored", "tailored", "applied"): where += " AND fit_score >= ?" params.append(min_score) diff --git a/src/applypilot/discovery/jobspy.py b/src/applypilot/discovery/jobspy.py index b5e54ff44..e9991a2b9 100644 --- a/src/applypilot/discovery/jobspy.py +++ b/src/applypilot/discovery/jobspy.py @@ -12,7 +12,12 @@ import time from datetime import datetime, timezone -from jobspy import scrape_jobs +try: + from jobspy import scrape_jobs + _JOBSPY_INSTALLED = True +except ImportError: + _JOBSPY_INSTALLED = False + scrape_jobs = None from applypilot import config from applypilot.database import get_connection, init_db, store_jobs @@ -300,6 +305,9 @@ def search_jobs( country_indeed: str = "usa", ) -> dict: """Run a single job search via JobSpy and store results in DB.""" + if not _JOBSPY_INSTALLED: + raise ImportError("python-jobspy is required for this module. Please install it with: pip install python-jobspy --no-deps") + if sites is None: sites = ["indeed", "linkedin", "zip_recruiter"] @@ -453,6 +461,9 @@ def run_discovery(cfg: dict | None = None) -> dict: Returns: Dict with stats: new, existing, errors, db_total, queries. """ + if not _JOBSPY_INSTALLED: + raise ImportError("python-jobspy is required for this module. Please install it with: pip install python-jobspy --no-deps") + if cfg is None: cfg = config.load_search_config() diff --git a/src/applypilot/discovery/smartextract.py b/src/applypilot/discovery/smartextract.py index cf49a9a2d..37fa4bff0 100644 --- a/src/applypilot/discovery/smartextract.py +++ b/src/applypilot/discovery/smartextract.py @@ -668,12 +668,14 @@ def extract_json(text: str) -> dict: return json.loads(text) except json.JSONDecodeError: pass - while text.endswith("}") or text.endswith("]"): + + match = re.search(r'([\{\[].*[\}\]])', text, re.DOTALL) + if match: try: - return json.loads(text) + return json.loads(match.group(1)) except json.JSONDecodeError: - text = text[:-1].rstrip() - raise json.JSONDecodeError("Could not parse JSON", text, 0) + pass + return None # -- JSON path resolution --------------------------------------------------- diff --git a/src/applypilot/scoring/tailor.py b/src/applypilot/scoring/tailor.py index 352fb5ff9..69197131f 100644 --- a/src/applypilot/scoring/tailor.py +++ b/src/applypilot/scoring/tailor.py @@ -272,21 +272,21 @@ def assemble_resume_text(data: dict, profile: dict) -> str: # Experience lines.append("EXPERIENCE") - for entry in data.get("experience", []): + for entry in data.get("experience") or []: lines.append(sanitize_text(entry.get("header", ""))) if entry.get("subtitle"): lines.append(sanitize_text(entry["subtitle"])) - for b in entry.get("bullets", []): + for b in entry.get("bullets") or []: lines.append(f"- {sanitize_text(b)}") lines.append("") # Projects lines.append("PROJECTS") - for entry in data.get("projects", []): + for entry in data.get("projects") or []: lines.append(sanitize_text(entry.get("header", ""))) if entry.get("subtitle"): lines.append(sanitize_text(entry["subtitle"])) - for b in entry.get("bullets", []): + for b in entry.get("bullets") or []: lines.append(f"- {sanitize_text(b)}") lines.append("") diff --git a/tests/test_apply_dry_run.py b/tests/test_apply_dry_run.py new file mode 100644 index 000000000..575d20c78 --- /dev/null +++ b/tests/test_apply_dry_run.py @@ -0,0 +1,36 @@ +"""Safety regressions for ApplyPilot dry-run mode.""" + +import unittest +from unittest.mock import patch + +from applypilot.apply import launcher + + +class TestApplyDryRun(unittest.TestCase): + @patch.object(launcher, "cleanup_worker") + @patch.object(launcher, "launch_chrome", return_value=object()) + @patch.object(launcher, "update_state") + @patch.object(launcher, "add_event") + @patch.object(launcher, "mark_result") + @patch.object(launcher, "release_lock") + @patch.object(launcher, "run_job", return_value=("applied", 100)) + @patch.object(launcher, "acquire_job") + def test_dry_run_never_marks_applied( + self, acquire_job, run_job, release_lock, mark_result, + add_event, update_state, launch_chrome, cleanup_worker, + ): + acquire_job.return_value = { + "url": "https://example.test/job/1", + "title": "Engineer", + "site": "Example", + } + + applied, failed = launcher.worker_loop(limit=1, dry_run=True) + + self.assertEqual((applied, failed), (0, 0)) + release_lock.assert_called_once_with("https://example.test/job/1") + mark_result.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_database.py b/tests/test_database.py new file mode 100644 index 000000000..f8bc5c3a1 --- /dev/null +++ b/tests/test_database.py @@ -0,0 +1,30 @@ +"""Unit tests for ApplyPilot database.py using unittest""" +import unittest +import tempfile +from pathlib import Path +from applypilot.database import get_connection, init_db, close_connection + +class TestDatabase(unittest.TestCase): + def test_init_db(self): + with tempfile.TemporaryDirectory() as tmp_dir: + db_file = Path(tmp_dir) / "test_applypilot.db" + conn = init_db(db_file) + self.assertIsNotNone(conn) + + cursor = conn.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='jobs'") + row = cursor.fetchone() + self.assertIsNotNone(row) + self.assertEqual(row[0], "jobs") + close_connection(db_file) + + def test_thread_local_connection(self): + with tempfile.TemporaryDirectory() as tmp_dir: + db_file = Path(tmp_dir) / "test_applypilot.db" + conn1 = get_connection(db_file) + conn2 = get_connection(db_file) + self.assertIs(conn1, conn2) + close_connection(db_file) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 000000000..610461e20 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,23 @@ +"""Unit tests for ApplyPilot pipeline.py using unittest""" +import unittest +from applypilot.pipeline import STAGE_ORDER, STAGE_META, _UPSTREAM + +class TestPipeline(unittest.TestCase): + def test_stage_order(self): + self.assertEqual(STAGE_ORDER, ("discover", "enrich", "score", "tailor", "cover", "pdf")) + + def test_stage_dependencies(self): + self.assertIsNone(_UPSTREAM["discover"]) + self.assertEqual(_UPSTREAM["enrich"], "discover") + self.assertEqual(_UPSTREAM["score"], "enrich") + self.assertEqual(_UPSTREAM["tailor"], "score") + self.assertEqual(_UPSTREAM["cover"], "tailor") + self.assertEqual(_UPSTREAM["pdf"], "cover") + + def test_stage_meta_completeness(self): + for stage in STAGE_ORDER: + self.assertIn(stage, STAGE_META) + self.assertIn("desc", STAGE_META[stage]) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_profile_safety.py b/tests/test_profile_safety.py new file mode 100644 index 000000000..555eaf74c --- /dev/null +++ b/tests/test_profile_safety.py @@ -0,0 +1,24 @@ +"""Production profile safety gates.""" + +import unittest + +from applypilot.config import profile_safety_reasons, validate_profile_for_application + + +class TestProfileSafety(unittest.TestCase): + def test_legacy_sample_profile_is_rejected(self): + profile = {"name": "Firstname Lastname (Sample Resume Candidate)", "email": "youremail@example.com"} + self.assertTrue(profile_safety_reasons(profile)) + with self.assertRaises(ValueError): + validate_profile_for_application(profile) + + def test_complete_structured_profile_is_allowed(self): + profile = { + "personal": {"full_name": "Real Person", "email": "person@domain.test"}, + "work_authorization": {}, "compensation": {}, "experience": {}, "resume_facts": {} + } + self.assertEqual(profile_safety_reasons(profile), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_validator.py b/tests/test_validator.py new file mode 100644 index 000000000..035827b72 --- /dev/null +++ b/tests/test_validator.py @@ -0,0 +1,59 @@ +"""Unit tests for ApplyPilot scoring/validator.py using unittest""" +import unittest +from applypilot.scoring.validator import sanitize_text, validate_json_fields + +class TestValidator(unittest.TestCase): + def test_sanitize_text(self): + raw = "Senior Engineer \u2014 Python \u2013 \u201cGreat\u201d \u2018Work\u2019" + cleaned = sanitize_text(raw) + self.assertEqual(cleaned, 'Senior Engineer, Python - "Great" \'Work\'') + + def test_validate_json_fields_missing_key(self): + data = {"title": "Software Engineer"} + profile = {} + result = validate_json_fields(data, profile) + self.assertFalse(result["passed"]) + self.assertTrue(any("Missing required field" in err for err in result["errors"])) + + def test_validate_json_fields_valid(self): + data = { + "title": "Backend Developer", + "summary": "Built distributed services using Python and PostgreSQL.", + "skills": {"Languages": "Python, SQL", "Tools": "Docker, Git"}, + "experience": [ + { + "header": "Acme Corp | Software Engineer", + "bullets": ["Optimized API latency by 40%."] + } + ], + "projects": [ + { + "header": "Open Source CLI", + "bullets": ["Developed file processing engine."] + } + ], + "education": "BS Computer Science" + } + profile = { + "resume_facts": {"preserved_companies": ["Acme Corp"]} + } + result = validate_json_fields(data, profile, mode="lenient") + self.assertTrue(result["passed"]) + self.assertEqual(len(result["errors"]), 0) + + def test_validate_json_fields_fabricated_skill(self): + data = { + "title": "Backend Developer", + "summary": "Experienced Rust developer", + "skills": {"Languages": "Python, Rust, Scala"}, + "experience": [{"header": "Acme Corp", "bullets": ["Built app"]}], + "projects": [{"header": "Proj", "bullets": ["Built app"]}], + "education": "BS CS" + } + profile = {"resume_facts": {"preserved_companies": ["Acme Corp"]}} + result = validate_json_fields(data, profile, mode="normal") + self.assertFalse(result["passed"]) + self.assertTrue(any("Fabricated skill" in err for err in result["errors"])) + +if __name__ == "__main__": + unittest.main()