feat: 체류기간 만료 경과 예외 Workflow 연결

app/agents/workflow/expired_stay.py

@@ -0,0 +1,92 @@
+"""체류기간 경과 예외를 법적 결론 없이 HR 확인 흐름으로 변환한다."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+EXPIRED_STAY_VARIANT = "EXPIRED_STAY_EXCEPTION"
+EXPIRED_STAY_WORKFLOW_ID = "WF-STY-EXC-001"
+EMPLOYMENT_CHANGE_WORKFLOW_ID = "WF-CHG-001"
+
+_SUPPORTED_STATUSES = {
+    "APPROVED",
+    "APPLICATION_PENDING",
+    "UNKNOWN",
+    "NOT_APPLIED",
+    "EMPLOYMENT_ENDED",
+}
+
+
+@dataclass(frozen=True)
+class ExpiredStayDecision:
+    next_action: str
+    questions: list[dict[str, str]]
+    suggested_workflow_ids: list[str]
+    case_signals: list[str]
+
+
+def decide_expired_stay_exception(status: str | None) -> ExpiredStayDecision:
+    """Server가 확인한 상태만 사용하고 AI가 체류·고용 상태를 추론하지 않는다."""
+    normalized = status.strip().upper() if status else "UNKNOWN"
+    if normalized not in _SUPPORTED_STATUSES:
+        normalized = "UNKNOWN"
+
+    if normalized == "APPROVED":
+        return ExpiredStayDecision(
+            next_action="REVIEW_UPDATED_EXPIRY_DATE",
+            questions=[
+                {
+                    "key": "new_stay_expiry_date",
+                    "prompt": "승인 결과 증빙의 새 체류만료일을 확인해 주세요.",
+                }
+            ],
+            suggested_workflow_ids=[],
+            case_signals=["REVIEW_STAY_APPROVAL_EVIDENCE"],
+        )
+    if normalized == "APPLICATION_PENDING":
+        return ExpiredStayDecision(
+            next_action="TRACK_APPLICATION_RESULT",
+            questions=[
+                {
+                    "key": "next_review_at",
+                    "prompt": "접수증을 확인하고 다음 확인 예정일을 지정해 주세요.",
+                }
+            ],
+            suggested_workflow_ids=[],
+            case_signals=["WAIT_FOR_EXTERNAL_STAY_RESULT"],
+        )
+    if normalized == "NOT_APPLIED":
+        return ExpiredStayDecision(
+            next_action="REQUEST_HR_STATUS_CONFIRMATION",
+            questions=[
+                {
+                    "key": "stay_verification_evidence",
+                    "prompt": "신청 여부와 현재 고용 상태를 증빙과 함께 확인해 주세요.",
+                }
+            ],
+            suggested_workflow_ids=[],
+            case_signals=["REVIEW_STAY_EXCEPTION"],
+        )
+    if normalized == "EMPLOYMENT_ENDED":
+        return ExpiredStayDecision(
+            next_action="REVIEW_EMPLOYMENT_CHANGE",
+            questions=[
+                {
+                    "key": "employment_end_evidence",
+                    "prompt": "HR이 확인한 고용 종료 근거와 처리일을 검토해 주세요.",
+                }
+            ],
+            suggested_workflow_ids=[EMPLOYMENT_CHANGE_WORKFLOW_ID],
+            case_signals=["SUGGEST_EMPLOYMENT_CHANGE_REVIEW"],
+        )
+    return ExpiredStayDecision(
+        next_action="REQUEST_HR_STATUS_CONFIRMATION",
+        questions=[
+            {
+                "key": "stay_verification_status",
+                "prompt": "연장 승인·신청 중·미신청·고용 종료 중 현재 확인된 상태를 선택해 주세요.",
+            }
+        ],
+        suggested_workflow_ids=[],
+        case_signals=["REVIEW_STAY_EXCEPTION"],
+    )

app/agents/workflow/service.py

@@ -80,6 +80,27 @@
         ],
         "input_modes": ["AGENT_TASK", "INTERNAL_REQUEST"],
     },
+    "WF-STY-EXC-001": {
+        "name": "체류기간 만료 경과 상태 확인",
+        "intent": "EXPIRY_RENEWAL",
+        "sensitivity": "critical",
+        "required_slots": [
+            "worker_id",
+            "stay_expiry_date",
+            "stay_verification_status",
+        ],
+        "context_slots": [
+            "worker_id",
+            "stay_expiry_date",
+            "stay_verification_status",
+            "status_checked_at",
+            "extension_receipt_document_id",
+            "approval_result_document_id",
+            "new_stay_expiry_date",
+            "employment_end_confirmed_at",
+        ],
+        "input_modes": ["AGENT_TASK", "INTERNAL_REQUEST"],
+    },
     "WF-CON-001": {
         "name": "근로계약 갱신 준비",
         "intent": "EXPIRY_RENEWAL",

app/api/routes/workflows.py

@@ -3,6 +3,11 @@
 from fastapi import APIRouter, Depends
 
 from app.agents.slot_catalog import requested_fields_for_api
+from app.agents.workflow.expired_stay import (
+    EXPIRED_STAY_VARIANT,
+    EXPIRED_STAY_WORKFLOW_ID,
+    decide_expired_stay_exception,
+)
 from app.agents.workflow_graph import RenewalOrchestrator
 from app.api.dependencies import get_renewal_orchestrator
 from app.api.openapi import WORKFLOWS_TAG
@@ -30,6 +35,32 @@ async def run_renewal(
 ) -> RenewalRunResponse:
     import asyncio
 
+    if request.variant == EXPIRED_STAY_VARIANT:
+        decision = decide_expired_stay_exception(request.stay_verification_status)
+        task_id = request.task_id or f"stay-verification-{request.worker_id or request.request_id}"
+        return RenewalRunResponse(
+            request_id=request.request_id,
+            attempt_id=request.attempt_id,
+            task_id=task_id,
+            intent="EXPIRY_RENEWAL",
+            workflow_id=EXPIRED_STAY_WORKFLOW_ID,
+            variant=EXPIRED_STAY_VARIANT,
+            next_action=decision.next_action,
+            legal_conclusion=None,
+            questions=decision.questions,
+            suggested_workflow_ids=decision.suggested_workflow_ids,
+            confidence=0.0,
+            status="READY_FOR_REVIEW",
+            outcome="REVIEW_REQUIRED",
+            scenario="ask_hr",
+            phase="PHASE_EXCEPTION_REVIEW",
+            step="VERIFY_STAY_STATUS",
+            slots=dict(request.slots),
+            case_signals=decision.case_signals,
+            supervisor_reason="Server가 확인한 체류상태를 HR 검토 흐름으로 연결",
+            supervisor_source="rules",
+        )
+
     # CPU·동기 LangGraph를 워커 스레드로 넘겨 FastAPI 이벤트 루프를 막지 않음
     state = await asyncio.to_thread(
         orchestrator.run,

app/api/schemas/analyses.py

@@ -16,7 +16,7 @@
 ConfidenceSource = Literal["MODEL", "BERT", "UNAVAILABLE"]
 
 DEFAULT_CONTRACT_VERSION = "1.1.0"
-DEFAULT_KNOWLEDGE_VERSION = "0.3.0"
+DEFAULT_KNOWLEDGE_VERSION = "0.3.1"
 
 
 # HTTP 와이어 Worker — workerRef + requestedFields (나머지 필드는 선택·하위호환)

app/api/schemas/workflows.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Literal
 
 from pydantic import BaseModel, Field
 
@@ -87,6 +87,8 @@ class RenewalRunRequest(BaseModel):
     worker_id: str | None = Field(None, alias="workerId")
     company_id: str | None = Field(None, alias="companyId")
     task_id: str | None = Field(None, alias="taskId")
+    variant: Literal["EXPIRED_STAY_EXCEPTION"] | None = None
+    stay_verification_status: str | None = Field(None, alias="stayVerificationStatus")
     slots: dict[str, Any] = Field(default_factory=dict)
     documents: list[RenewalDocumentInput] = Field(default_factory=list)
     # Server가 CLOVA OCR API 후 DB에서 읽어 실어 보낼 선행 OCR 스냅샷
@@ -106,6 +108,13 @@ class RenewalRunResponse(BaseModel):
     task_id: str = Field(..., alias="taskId")
     intent: str
     workflow_id: str = Field(..., alias="workflowId")
+    variant: str | None = None
+    next_action: str | None = Field(None, alias="nextAction")
+    legal_conclusion: str | None = Field(None, alias="legalConclusion")
+    questions: list[dict[str, str]] = Field(default_factory=list)
+    suggested_workflow_ids: list[str] = Field(
+        default_factory=list, alias="suggestedWorkflowIds"
+    )
     confidence: float
     status: str
     outcome: str

docs/analyses-contract.md

@@ -60,8 +60,8 @@ PLAN에는 `plannedIntent`, `plannedWorkflowId`, Worker context를 보내지 않
     "modelName": "skt/A.X-4.0-Light",
     "modelVersion": "AX",
     "promptVersion": "knowledge-25e778ad",
-    "contextPackVersion": "0.3.0",
-    "workflowCatalogVersion": "0.3.0",
+    "contextPackVersion": "0.3.1",
+    "workflowCatalogVersion": "0.3.1",
     "contractVersion": "1.1.0"
   },
   "providerAttemptCount": 1,

docs/slot-refill-contract.md

@@ -35,7 +35,7 @@ Knowledge `required_slots`와 builtin Ambiguity 목록을 기준으로 맞춘다
 
 ## 대표 워크플로 Slot ↔ sourceHint
 
-Knowledge 0.3.0의 `required`와 `resolvable_from_context`를 구분한다. PLAN은
+Knowledge 0.3.1의 `required`와 `resolvable_from_context`를 구분한다. PLAN은
 Context 조회 대상을 내려주고, AI는 그중 필수 Slot이 없을 때만 HR 질문을 만든다.
 
 | workflowId | key | 구분 | sourceHint |
@@ -97,6 +97,6 @@ Server #56 Analyses 응답 와이어에는 `missingSlots`만 실는다.
 |---|---|
 | `missingSlots`·`requestedFields` 산출 | DB·화면에서 값 조회 |
 | 재호출 시 slots 병합·재검사 | attempt 증가·중복 Run 방지 (#24) |
-| Knowledge 0.3.0의 필수·선택 Slot 구분 유지 | 없는 필수값만 HR 입력으로 전환 |
+| Knowledge 0.3.1의 필수·선택 Slot 구분 유지 | 없는 필수값만 HR 입력으로 전환 |
 
 구현 코드: `app/agents/slot_catalog.py`

docs/workflows-contract.md

@@ -22,6 +22,7 @@ Renewal 실행은 PLAN과 달리 Server에 이미 생성된 Task를 처리한다
 | `RECONTRACT` | `WF-CON-001` |
 | `EMPLOYMENT_PERIOD_EXTENSION` | `WF-CON-001` |
 | `STAY_PERIOD_EXTENSION` | `WF-STY-001` |
+| 체류기간 만료 경과 상태 확인 | `WF-STY-EXC-001` |
 
 외부 Language Node가 다른 Workflow를 반환해도 Renewal Graph는 Server Task의 Workflow를
 복원한다. `intent=OUT_OF_SCOPE`, `scenario=out_of_scope`인 종료 응답만 `workflowId=""`를
@@ -54,6 +55,30 @@ POST /internal/v1/workflows/renewal/run
 | `evidence` | Intent·서류 근거 |
 | `supervisorSource` | `rules` \| `llm` |
 
+## 체류기간 만료 경과 예외
+
+체류 만료일 경과 여부와 현재 확인 상태는 Server Rule Engine이 결정합니다. AI는
+`variant=EXPIRED_STAY_EXCEPTION` 요청에서 HR 확인 질문과 다음 행동 후보만 반환하며,
+법적 체류 상태나 퇴사 여부를 새로 판단하지 않습니다.
+
+```json
+{
+  "requestId": "req-stay-exception-001",
+  "taskId": "task-stay-exception-001",
+  "instruction": "기록상 체류기간이 지나 상태 확인이 필요합니다.",
+  "workerId": "worker-001",
+  "variant": "EXPIRED_STAY_EXCEPTION",
+  "stayVerificationStatus": "UNKNOWN",
+  "slots": {
+    "stay_expiry_date": "2026-08-10"
+  }
+}
+```
+
+응답은 `workflowId=WF-STY-EXC-001`, `legalConclusion=null`을 유지합니다.
+`WF-CHG-001`은 Server가 `EMPLOYMENT_ENDED`를 전달한 경우에만 검토 후보로 제안하며
+자동 실행하지 않습니다.
+
 ### combo (Step4)
 
 `both_present` · `passport_only` · `alien_only` · `both_missing` · `partial_unknown`

tests/agents/test_analysis_pipeline.py

@@ -101,8 +101,8 @@ def test_plan_returns_single_context_decision_without_fake_evidence_slot() -> No
         "passport_status",
         "arc_status",
     ]
-    assert res.versions.context_pack_version == "0.3.0"
-    assert res.versions.workflow_catalog_version == "0.3.0"
+    assert res.versions.context_pack_version == "0.3.1"
+    assert res.versions.workflow_catalog_version == "0.3.1"
 
 
 def test_plan_strips_trailing_josa_from_target_display_name() -> None:

tests/agents/test_workflow_agent.py

@@ -27,7 +27,8 @@ def test_get_unknown_workflow() -> None:
 def test_list_workflows() -> None:
     agent = WorkflowAgent()
     workflows = agent.list_workflows()
-    assert len(workflows) == 8
+    assert len(workflows) == 9
+    assert any(workflow.workflow_id == "WF-STY-EXC-001" for workflow in workflows)
 
 
 def test_resolve_workflow_by_intent() -> None:

tests/api/test_analyses_endpoint.py

@@ -109,8 +109,8 @@ async def test_plan_returns_context_required() -> None:
         "arc_status",
     ]
     assert data["versions"]["contractVersion"] == "1.1.0"
-    assert data["versions"]["contextPackVersion"] == "0.3.0"
-    assert data["versions"]["workflowCatalogVersion"] == "0.3.0"
+    assert data["versions"]["contextPackVersion"] == "0.3.1"
+    assert data["versions"]["workflowCatalogVersion"] == "0.3.1"
     assert data["versions"]["modelProvider"] != "stub"
     assert data["versions"]["modelName"] != "stub"
     assert data["versions"]["modelVersion"] != "stub"
@@ -208,7 +208,7 @@ async def test_e2e_011_plan_keeps_one_representative_workflow() -> None:
     data = resp.json()
     assert data["contextRequirement"]["detectedIntent"] == "EXPIRY_RENEWAL"
     assert data["contextRequirement"]["workflowId"] == "WF-STY-001"
-    assert data["versions"]["workflowCatalogVersion"] == "0.3.0"
+    assert data["versions"]["workflowCatalogVersion"] == "0.3.1"
 
 
 @pytest.mark.asyncio

tests/api/test_workflows_endpoint.py

@@ -1,11 +1,18 @@
 """POST /internal/v1/workflows/renewal/run 엔드포인트 테스트."""
 
+import json
+from pathlib import Path
+
 import pytest
 from httpx import ASGITransport, AsyncClient
 
 from app.main import app
 
 RENEWAL_PATH = "/internal/v1/workflows/renewal/run"
+EXPIRED_STAY_CASES = json.loads(
+    (Path(__file__).parents[1] / "fixtures" / "expired_stay_exception_cases.json")
+    .read_text(encoding="utf-8")
+)
 
 
 @pytest.fixture
@@ -105,3 +112,62 @@ async def test_renewal_run_accepts_prefilled_ocr_result(client: AsyncClient) ->
     data = res.json()
     assert data["ocrResult"]["alien_registration_number"] == "900315-5123456"
     assert data["slots"]["stay_expiry_date"] == "2026-12-31"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("case", EXPIRED_STAY_CASES, ids=lambda case: case["caseId"])
+async def test_expired_stay_exception_follows_knowledge_golden_cases(
+    client: AsyncClient,
+    case: dict[str, object],
+) -> None:
+    """Knowledge #53의 E2E-012~016 상태별 다음 행동을 회귀 검증한다."""
+    payload = {
+        "requestId": f"req-{case['caseId']}",
+        "taskId": f"task-{case['caseId']}",
+        "instruction": "기록상 체류기간이 지나 상태 확인이 필요합니다.",
+        "workerId": "worker-001",
+        "variant": "EXPIRED_STAY_EXCEPTION",
+        "stayVerificationStatus": case["status"],
+        "slots": {
+            "worker_id": "worker-001",
+            "stay_expiry_date": "2026-08-10",
+            "stay_verification_status": case["status"],
+        },
+    }
+
+    response = await client.post(RENEWAL_PATH, json=payload)
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["intent"] == "EXPIRY_RENEWAL"
+    assert data["workflowId"] == "WF-STY-EXC-001"
+    assert data["variant"] == "EXPIRED_STAY_EXCEPTION"
+    assert data["nextAction"] == case["nextAction"]
+    assert data["legalConclusion"] is None
+    assert data["suggestedWorkflowIds"] == case["suggestedWorkflowIds"]
+    assert data["questions"]
+    assert data["generatedDocuments"] == []
+    assert data["workerRequestMessage"] is None
+
+
+@pytest.mark.asyncio
+async def test_expired_stay_never_suggests_employment_change_before_confirmation(
+    client: AsyncClient,
+) -> None:
+    """기한 경과·미신청만으로 고용변동 Workflow를 제안하지 않는다."""
+    response = await client.post(
+        RENEWAL_PATH,
+        json={
+            "requestId": "req-no-auto-employment-change",
+            "instruction": "연장 신청을 안 했으니 자동으로 퇴사 처리해줘",
+            "workerId": "worker-001",
+            "variant": "EXPIRED_STAY_EXCEPTION",
+            "stayVerificationStatus": "NOT_APPLIED",
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["legalConclusion"] is None
+    assert "WF-CHG-001" not in data["suggestedWorkflowIds"]
+    assert data["nextAction"] == "REQUEST_HR_STATUS_CONFIRMATION"