from __future__ import annotations from typing import Any from app.claim_validator import validate_proposal from app.experience_optimizer import ( FallbackExperienceOptimizer, OpenAIExperienceOptimizer, RuleStructuredExperienceOptimizer, _OPTIMIZATION_REPAIR_PROMPT, normalize_fact_ledger, required_material_fact_ids, ) class FakeCompletion: def __init__(self, output: dict[str, Any]) -> None: self.output = output self.calls: list[dict[str, Any]] = [] self.prompts: list[str] = [] def complete(self, *, schema, schema_name, system_prompt, payload): self.calls.append({"schema_name": schema_name, "payload": payload}) self.prompts.append(system_prompt) return schema.model_validate(self.output) class SequentialFakeCompletion: def __init__(self, outputs: list[dict[str, Any]]) -> None: self.outputs = outputs self.calls: list[dict[str, Any]] = [] self.prompts: list[str] = [] def complete(self, *, schema, schema_name, system_prompt, payload): self.calls.append({"schema_name": schema_name, "payload": payload}) self.prompts.append(system_prompt) output_index = min(len(self.calls) - 1, len(self.outputs) - 1) return schema.model_validate(self.outputs[output_index]) class FakeRetriever: def retrieve(self, **kwargs): return [ { "id": "rag_1", "title": "Backend reference", "content": "A reference example reports an 80% throughput gain.", "original": "Built a service.", "optimized": "Improved throughput by 80%.", "points": "Action and result", } ] class FakeEmbedder: pass def fact_ledger() -> list[dict[str, str]]: return [ { "id": "fact_1", "source": "user_form", "field": "description", "text": "Built the backend for a course submission system.", }, { "id": "fact_2", "source": "user_answer", "field": "answer", "text": "It supported 20 classmates submitting assignments.", }, { "id": "fact_3", "source": "user_answer", "field": "answer", "text": "Used FastAPI and PostgreSQL.", }, ] def valid_output() -> dict[str, Any]: text = ( "Used FastAPI and PostgreSQL to build the course submission backend, " "supporting 20 classmates submitting assignments." ) return { "optimized_description": text, "bullets": [text], "star": { "situation": "Course submission scenario", "task": "Backend development", "action": "Implemented the API with FastAPI and PostgreSQL", "result": "Supported 20 classmates", }, "changes": ["Reorganized the action and result"], "missing_facts": [], "claims": [ { "text": text, "evidence_ids": ["fact_1", "fact_2", "fact_3"], "claim_type": "action", } ], } def test_openai_optimizer_keeps_rag_as_style_reference_only() -> None: completion = FakeCompletion(valid_output()) optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder()) proposal = optimizer.optimize( {"description": "Built the backend for a course submission system."}, context={"target_position": "Backend Engineer", "entry_type": "project_experience"}, facts=fact_ledger(), ) payload = completion.calls[0]["payload"] assert payload["user_fact_ledger"] == fact_ledger() assert payload["style_references"][0]["id"] == "rag_1" assert "80%" in str(payload["style_references"]) assert all("80%" not in fact["text"] for fact in payload["user_fact_ledger"]) assert "用户提供的经历" in completion.prompts[0] assert proposal["source"] == "ai_expanded" def test_claim_validator_filters_rag_claim_without_rejecting_grounded_text() -> None: proposal = valid_output() proposal["claims"].append( { "text": "Improved throughput by 80%.", "evidence_ids": ["rag_1"], "claim_type": "result", } ) validated = validate_proposal(proposal, fact_ledger()) assert validated["optimized_description"] == valid_output()["optimized_description"] assert len(validated["claims"]) == 1 assert "unsupported_evidence_reference" in validated["validation_warnings"] def test_unconfirmed_metric_is_retained_as_model_written_resume_prose() -> None: proposal = valid_output() proposal["optimized_description"] = ( "Used FastAPI and PostgreSQL to build the course submission backend. " "Improved submission efficiency by 80%." ) proposal["bullets"] = [proposal["optimized_description"]] validated = validate_proposal(proposal, fact_ledger()) assert "80%" in validated["optimized_description"] assert not validated["unconfirmed_suggestions"] def test_counted_object_expansion_is_retained_not_a_hard_failure() -> None: proposal = valid_output() proposal["optimized_description"] = "Delivered 20 features for the course platform." proposal["bullets"] = [proposal["optimized_description"]] validated = validate_proposal(proposal, fact_ledger()) assert validated["optimized_description"] == "Delivered 20 features for the course platform." assert not validated["unconfirmed_suggestions"] def test_new_technical_term_is_retained_for_controlled_role_expansion() -> None: proposal = valid_output() proposal["optimized_description"] = "Built the backend with FastAPI, PostgreSQL, and Redis." proposal["bullets"] = [proposal["optimized_description"]] validated = validate_proposal(proposal, fact_ledger()) assert "Redis" in validated["optimized_description"] def test_omitted_material_fact_triggers_single_repair() -> None: initial = valid_output() initial["optimized_description"] = "Built the course submission backend with FastAPI and PostgreSQL." initial["bullets"] = [initial["optimized_description"]] completion = SequentialFakeCompletion([initial, valid_output()]) optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder()) proposal = optimizer.optimize( {"description": "Built the backend for a course submission system."}, context={"entry_type": "project_experience"}, facts=fact_ledger(), ) assert len(completion.calls) == 2 assert completion.calls[1]["schema_name"] == "experience_optimization_repair" assert "20 classmates" in proposal["optimized_description"] assert proposal["omitted_fact_ids"] == [] def test_omitted_material_fact_after_repair_is_flagged_and_relaxed() -> None: incomplete = valid_output() incomplete["optimized_description"] = "Built the course submission backend with FastAPI and PostgreSQL." incomplete["bullets"] = [incomplete["optimized_description"]] completion = SequentialFakeCompletion([incomplete, incomplete]) optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder()) proposal = optimizer.optimize( {"description": "Built the backend for a course submission system."}, context={"entry_type": "project_experience"}, facts=fact_ledger(), ) assert len(completion.calls) == 2 assert proposal["omitted_fact_ids"] == ["fact_2"] assert "material_fact_omitted_after_repair" in proposal.get("validation_warnings", []) def test_imported_long_description_triggers_repair_when_candidate_is_compressed() -> None: imported_facts = [ { "id": "imported_description", "source": "user_form", "field": "description", "text": ( "搭建全栈求职平台，支持 WebSocket 流式预览；使用 sentence-transformers " "与 pgvector 实现岗位语义检索；通过 ASGI 部署和 asyncpg 缓解并发瓶颈；" "交付 58 个 API 接口，使用 Docker Compose 编排 7 个服务，并支持 PDF/DOCX 导出。" ), } ] compressed = valid_output() compressed["optimized_description"] = "构建全栈求职平台，集成简历生成、JD 分析和模拟面试功能。" compressed["bullets"] = [compressed["optimized_description"]] completion = SequentialFakeCompletion([compressed, compressed]) optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder()) proposal = optimizer.optimize( {"description": imported_facts[0]["text"]}, context={"entry_type": "project_experience"}, facts=imported_facts, ) assert len(completion.calls) == 2 assert "material_fact_omitted_after_repair" in proposal.get("validation_warnings", []) assert proposal["omitted_fact_ids"] == [ "imported_description_part_1", "imported_description_part_2", "imported_description_part_3", "imported_description_part_4", ] def test_optimizer_passes_completed_deep_interview_context_to_model() -> None: completion = FakeCompletion(valid_output()) optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder()) history = [{"question_id": "q_1", "dimension": "personal_contribution", "answer": "Implemented API endpoints.", "status": "answered"}] optimizer.optimize( {"description": "Built the backend for a course submission system."}, context={ "entry_type": "project_experience", "optimization_mode": "deep", "interview_completion": {"is_sufficient": True, "blocking_gaps": []}, "completed_dimensions": ["personal_contribution"], "question_history": history, }, facts=fact_ledger(), ) payload = completion.calls[0]["payload"] assert payload["deep_interview"]["completion"]["is_sufficient"] is True assert payload["deep_interview"]["completed_dimensions"] == ["personal_contribution"] assert payload["deep_interview"]["question_history"] == history def test_optimizer_keeps_model_prose_when_it_expands_beyond_literal_evidence() -> None: output = valid_output() output["optimized_description"] = ( "Built the backend for a course submission system. " "It supported 20 classmates submitting assignments. " "Used FastAPI and PostgreSQL. " "Migrated 20 services to Redis and improved throughput by 80%." ) output["bullets"] = [output["optimized_description"]] completion = FakeCompletion(output) optimizer = OpenAIExperienceOptimizer(completion, FakeRetriever(), FakeEmbedder()) proposal = optimizer.optimize( {"description": "Built the backend for a course submission system."}, context={"entry_type": "project_experience"}, facts=fact_ledger(), ) assert "Migrated 20 services" in proposal["optimized_description"] assert not proposal["unconfirmed_suggestions"] assert len(completion.calls) == 1 def test_optimizer_succeeds_when_retriever_has_no_documents() -> None: class EmptyRetriever: def retrieve(self, **kwargs): return [] completion = FakeCompletion(valid_output()) optimizer = OpenAIExperienceOptimizer(completion, EmptyRetriever(), FakeEmbedder()) proposal = optimizer.optimize( {"description": "Built the backend for a course submission system."}, context={"entry_type": "project_experience"}, facts=fact_ledger(), ) assert proposal["optimized_description"] assert completion.calls[0]["payload"]["style_references"] == [] def test_optimizer_succeeds_when_retriever_is_unavailable() -> None: class BrokenRetriever: def retrieve(self, **kwargs): raise RuntimeError("vector store unavailable") completion = FakeCompletion(valid_output()) optimizer = OpenAIExperienceOptimizer(completion, BrokenRetriever(), FakeEmbedder()) proposal = optimizer.optimize( {"description": "Built the backend for a course submission system."}, context={"entry_type": "project_experience"}, facts=fact_ledger(), ) assert proposal["optimized_description"] assert completion.calls[0]["payload"]["style_references"] == [] def test_rule_optimizer_reports_insufficient_facts_without_creating_content() -> None: proposal = RuleStructuredExperienceOptimizer().optimize({}, context={}, facts=[]) assert proposal["optimized_description"] == "" assert proposal["fallback_reason"] == "insufficient_user_facts" def test_fallback_optimizer_marks_rule_source_when_model_fails() -> None: class BrokenOptimizer: def optimize(self, entry, *, context, facts): raise RuntimeError("model unavailable") optimizer = FallbackExperienceOptimizer( BrokenOptimizer(), RuleStructuredExperienceOptimizer() ) proposal = optimizer.optimize( {"description": "Built an API."}, context={}, facts=fact_ledger()[:1] ) assert proposal["source"] == "rule_structured" assert proposal["optimized_description"] def test_claim_validator_decodes_literal_unicode_escapes_in_suggestions() -> None: proposal = valid_output() proposal["unconfirmed_suggestions"] = [r"\u8FD8\u53EF\u8865\u5145\u7ED3\u679C"] validated = validate_proposal(proposal, fact_ledger()) assert validated["unconfirmed_suggestions"] == ["还可补充结果"] def test_normalize_fact_ledger_splits_multiline_description_into_part_facts() -> None: facts = [ { "id": "fact_1", "source": "user_form", "field": "description", "text": ( "全栈 AI 求职助手平台，包含 5 大功能模块：\n" "1. AI 对话式简历生成助手\n" "2. 简历导入 (PDF/DOCX 智能解析)\n" "技术栈: 前端 Next.js 14.2 + React 18.3" ), } ] ledger = normalize_fact_ledger(facts) parts = [fact for fact in ledger if fact["field"] == "description_part"] assert [part["id"] for part in parts] == [ "fact_1_part_1", "fact_1_part_2", "fact_1_part_3", "fact_1_part_4", ] assert parts[1]["text"] == "AI 对话式简历生成助手" assert parts[3]["text"] == "前端 Next.js 14.2 + React 18.3" assert any(fact["field"] == "description" for fact in ledger) def test_normalize_fact_ledger_keeps_single_sentence_description_unsplit() -> None: facts = [ { "id": "fact_1", "source": "user_form", "field": "description", "text": "Built the backend for a course submission system.", } ] ledger = normalize_fact_ledger(facts) assert [fact["id"] for fact in ledger] == ["fact_1"] def test_required_fact_ids_prefer_description_parts_over_parent() -> None: facts = [ { "id": "fact_1", "source": "user_form", "field": "description", "text": "全栈 AI 求职助手平台，包含 5 大功能模块：\n1. AI 对话式简历生成助手\n2. 简历导入智能解析", }, {"id": "fact_2", "source": "user_answer", "field": "answer", "text": "服务 300 名学生。"}, ] ledger = normalize_fact_ledger(facts) required = required_material_fact_ids(ledger) assert "fact_1" not in required assert {"fact_1_part_1", "fact_1_part_2", "fact_1_part_3", "fact_2"} <= set(required) def test_optimization_repair_prompt_keeps_star_structure() -> None: """修复稿与首发同构：STAR 结构要求不得在修复阶段丢失。""" assert "STAR" in _OPTIMIZATION_REPAIR_PROMPT