계획 패턴 - 작업 목록과 상태 파일

이 장에서 배우는 것

《AI 에이전트 활용과 하네스 설계》의 이번 장은 ‘계획 패턴 - 작업 목록과 상태 파일’이다. 앞 장에서 도구의 이름과 입력, 출력, 위험도를 정했다. 이제 그 도구들을 어떤 순서로 실행하고, 어디까지 끝났는지 어떻게 남길지 정한다. 도구가 각각 잘 작동하더라도 작업 순서와 완료 기준이 없으면 전체 목표가 끝났다고 판단하기 어렵다.

계획은 앞으로 할 일을 적은 문서이면서 실행기가 읽는 입력이다. 상태는 실제로 확인한 진행 상황이다. 둘을 구분하면 중간에 멈춘 작업을 이어서 처리할 수 있고, 실행 중 발견한 조건을 계획에 반영할 수도 있다. 이번 예제는 규칙 기반 가짜 모델이 만든 계획을 JSON 파일로 저장한 뒤, 일부 작업을 실행하고 멈춘다. 이어서 계획을 수정하고 상태 파일을 읽어 남은 작업을 끝낸다.

• 목표를 입력과 산출물이 분명한 작업 목록으로 나눈다.
• 각 작업에 실행 방법과 완료 확인 방법을 함께 붙인다.
• 계획 파일과 상태 파일을 구분하고, 중단 후 완료 작업을 확인하며 재시작한다.
• 끝난 작업의 의미를 보존하면서 남은 계획을 수정한다.
• 가짜 모델의 계획과 실행 결과를 코드로 검사한다.

문제 상황

작은 문서 정리 프로그램을 만든다고 가정한다. 사용자가 원하는 결과는 “여러 메모에서 공백과 빈 항목을 정리하고, 번호가 붙은 목록을 만든다”이다. 필요한 도구는 이미 있다. 메모를 가져오는 도구, 문자열을 정리하는 도구, 목록을 만드는 도구다. 처음에는 세 도구를 차례대로 호출하면 충분해 보인다.

그런데 문자열 정리까지 끝난 순간 프로그램을 중단해야 한다면 문제가 생긴다. 다시 시작할 때 메모를 처음부터 가져올지, 정리된 파일을 사용할지 판단할 근거가 필요하다. 파일이 존재한다는 사실만으로는 정리가 끝났는지 알 수 없다. 실행 도중 만들어진 파일일 수도 있고, 이전 실행의 결과일 수도 있다.

계획도 고정되어 있지 않다. 정리된 항목을 확인한 뒤, 목록을 만들기 전에 항목 수를 별도로 확인하자는 요구가 생길 수 있다. 이때 전체 계획을 새로 만들고 진행 상태를 초기화하면 이미 끝난 일을 다시 한다. 반대로 기존 상태를 아무 검사 없이 새 계획에 붙이면 다른 의미의 작업을 완료한 것으로 취급할 수 있다.

이번 장에서는 이 상황을 작은 하네스로 재현한다. 실제 사용자 파일은 읽지 않는다. 프로그램 안의 메모를 임시 폴더에 저장해 사용하고, 실행이 끝나면 폴더를 지운다. 출력에는 임시 경로나 현재 시각을 넣지 않으므로 실행할 때마다 같은 결과를 얻는다. 여기서 만드는 재시작은 같은 임시 폴더를 유지한 채 실행 함수를 다시 호출하는 방식이다. 별도 프로세스로 재시작하는 경우에도 계획과 상태를 읽는 원리는 같지만, 임시 폴더의 수명은 따로 정해야 한다.

목표를 확인 가능한 작업으로 나누기

작업을 나눌 때 기준은 문장의 길이가 아니다. 입력을 받아 어떤 산출물을 만들고, 그 산출물을 어떻게 확인할 수 있는지가 기준이다. “메모를 잘 정리한다”는 목표로는 유용하지만 실행 단위로는 모호하다. “원본 배열을 읽어 양끝 공백을 제거하고 빈 문자열을 제외한 배열을 저장한다”는 작업은 입력과 결과를 확인할 수 있다.

이번 예제의 가짜 모델은 목표 문자열을 받아 정해진 작업 목록을 반환한다. 모델이 실행 코드를 작성하는 방식은 사용하지 않는다. 작업에는 미리 등록한 도구 이름과 완료 확인 방법의 이름만 들어간다. 실행기는 그 이름을 허용 목록과 대조한다. 계획이 문법상 올바른 JSON이라는 사실과, 실행할 수 있는 계획이라는 사실은 서로 다르다.

작업은 산출물과 완료 확인 방법을 함께 가진다

작업 식별자	실행할 일	산출물	완료 확인
collect	예제 메모 가져오기	raw.json	고정된 원본 배열과 일치
normalize	공백과 빈 항목 정리	clean.json	원본에서 계산한 정리 결과와 일치
audit	정리된 항목 수 기록	audit.json	저장된 수와 실제 항목 수가 일치
report	번호를 붙인 목록 만들기	report.txt	정리된 배열에서 만든 목록과 일치

처음 계획에는 collect, normalize, report만 들어간다. audit는 중단 뒤 계획을 고칠 때 추가한다. 모든 작업은 앞에서 만든 산출물만 읽는다. 별도의 의존 관계 자료구조를 만들지 않고 목록 순서로 관계를 나타낸다. 이 방식은 작은 순차 작업에 적합하다. 동시에 실행할 수 있는 작업이나 여러 갈래로 나뉘는 작업은 이 예제의 범위에 포함하지 않는다.

식별자는 작업의 위치와 구별해야 한다. 세 번째 위치에 새 작업이 들어오면 기존 세 번째 작업은 네 번째가 된다. 그러나 report라는 식별자는 그대로다. 상태를 배열 위치로 저장하면 계획 삽입 때 상태의 주인이 바뀔 수 있다. 따라서 상태 파일은 식별자를 키로 사용한다.

완료 확인은 실행 함수의 반환값과도 구별한다. 도구가 오류 없이 끝났다는 것은 호출이 종료되었다는 뜻이다. 목표에 맞는 파일을 만들었다는 뜻까지 포함하지는 않는다. 예제에서는 도구 실행 뒤 파일을 다시 읽고 기대 결과와 비교한다. 파일 이름이 맞더라도 내용이 다르면 완료 상태로 기록하지 않는다.

비교에 쓰는 기대 결과는 예제의 명세에서 나온다. 정리 작업은 양끝 공백을 제거하고 빈 항목을 제외하되 중복은 유지한다. 입력에 같은 메모가 두 번 있으면 결과에도 두 번 있어야 한다. 명세에 없는 중복 제거를 임의로 추가하면 더 정돈된 결과처럼 보이더라도 완료 확인에서 실패한다.

계획 파일과 상태 파일의 역할 나누기

계획 파일은 무엇을 할지 정한다. 상태 파일은 그중 무엇을 확인했는지 기록한다. 계획에는 버전과 작업 목록을 넣고, 각 작업에는 식별자, 도구, 완료 확인 방법을 넣는다. 상태에는 연결된 계획 버전과 작업별 진행 상태를 넣는다. 계획을 고칠 때 버전을 올리면 어느 계획에 맞춰 상태를 정리했는지 알 수 있다.

계획과 상태는 서로 다른 질문에 답한다

파일	담는 내용	답하는 질문
plan.json	버전, 작업 식별자, 도구, 확인 방법	무엇을 어떤 순서로 할 것인가
state.json	계획 버전, 작업 정의, 작업별 상태	어떤 정의의 작업을 어디까지 확인했는가
산출물 파일	원본, 정리 결과, 수량, 최종 목록	실행 결과가 실제로 남아 있는가

진행 상태는 대기(pending), 실행 중(running), 완료(done) 세 가지로 둔다. 실행 직전에 running을 저장한다. 도구를 실행하고 완료 확인을 통과한 뒤 done을 저장한다. 상태 파일을 읽었을 때 running이 남아 있다면 도구 실행과 완료 기록 사이에서 멈췄을 수 있다. 이 예제는 해당 작업을 pending으로 돌려 다시 실행한다.

이 정책이 가능한 이유는 도구가 정해진 파일을 같은 내용으로 덮어쓰기 때문이다. 같은 입력으로 여러 번 실행해도 결과가 같다는 성질을 멱등성(idempotency)이라 한다. 메모를 가져오는 도구는 고정 배열을 다시 저장하고, 정리 도구는 원본에서 다시 계산한다. 재실행으로 항목이 누적되지 않는다. 다른 시스템에 알림을 보내거나 금액을 처리하는 도구에는 같은 정책을 그대로 적용할 수 없다.

done도 무조건 신뢰하지 않는다. 재시작할 때 완료 표시가 붙은 작업의 산출물을 다시 확인한다. 파일이 없어졌거나 내용이 바뀌었다면 예외를 내고 멈춘다. 이번 예제에서는 그때 어떤 작업까지 되돌릴지 자동으로 정하지 않는다. 잘못된 완료 표시를 조용히 통과시키는 일을 막는 데 초점을 둔다.

상태 저장은 임시 이름으로 JSON을 먼저 쓰고, 쓰기가 끝나면 기존 상태 파일을 교체한다. 같은 폴더 안에서 교체하므로 새 내용을 쓰는 도중 기존 파일을 직접 잘라내지 않는다. 다만 이 작은 구현은 전원 손실 뒤 저장 내용의 유지까지 보장하지 않는다. 여러 실행기가 동시에 같은 상태를 수정하는 경우도 다루지 않는다. 예제의 전제는 실행기가 하나라는 것이다.

재시작 기능을 확인하려면 같은 실행 함수가 이전 호출의 지역 변수를 참조하지 않아야 한다. 아래 run 함수는 호출될 때마다 계획 파일과 상태 파일을 새로 읽는다. 첫 호출에서 만든 파이썬 딕셔너리를 두 번째 호출에 넘기지 않는다. 따라서 이어서 처리할 위치의 근거는 메모리가 아니라 파일에 있다.

실행 중 계획을 고치는 기준

계획을 수정하는 이유는 새 입력이나 확인 결과가 목표를 달성하는 방법에 영향을 주기 때문이다. 계획을 세웠다는 이유만으로 끝까지 같은 목록을 고집할 필요는 없다. 반대로 실행이 마음에 들지 않을 때마다 전체 목록을 바꾸면 진행 상태가 무엇을 뜻하는지 알기 어려워진다. 변경 범위와 완료 작업의 연결을 함께 검사해야 한다.

이번 예제는 처음 두 작업을 마친 뒤, 최종 목록 전에 항목 수를 기록하는 audit를 추가한다. 목표는 유지하고 남은 절차만 보완하는 변경이다. 가짜 모델은 현재 계획과 수정 요청을 받아 새 목록을 반환한다. 실제 모델을 사용하는 상황에서 요청문은 다음처럼 쓸 수 있다.

현재 완료한 collect와 normalize의 식별자, 도구, 완료 확인 방법을 유지하라. report 바로 앞에 정리된 항목 수를 기록하는 audit 작업을 추가하라. 계획 버전은 하나 올려라.

이 요청이 잘 쓰였다고 해서 결과를 바로 실행하지 않는다. 실행기는 완료한 작업들이 여전히 목록의 앞부분에 같은 순서로 있는지 검사한다. 또한 이전에 저장한 작업 정의와 새 정의를 비교한다. 완료한 normalize의 도구나 확인 방법이 달라지면 변경을 거부한다. 같은 식별자에 다른 일을 붙이지 못하게 하는 것이다.

아직 끝나지 않은 작업은 새 계획의 정의를 적용하고 pending으로 둔다. 이번 구현은 완료 작업이 앞쪽에 연속해서 놓이는 순차 실행을 전제로 한다. 완료 작업 앞에 새 작업을 삽입하거나 완료 작업을 지우는 변경은 허용하지 않는다. 필요한 변경이라면 이미 만든 결과를 무효화하는 규칙부터 별도로 정해야 한다.

계획 수정의 검토에서는 차이(diff)를 읽는다. 이 예제의 차이는 버전 증가와 audit 추가이며, 나머지 정의는 그대로다. 출력은 작업 순서의 전후를 보여 주고, 코드는 기존 작업 정의가 달라지지 않았는지도 검사한다. 실제 작업에서도 AI가 제안한 계획이나 코드를 실행하고 테스트한 뒤, 의도한 변경만 들어갔는지 차이를 검토해야 한다.

완성 코드

다음 내용을 main.py로 저장한다. 프로그램은 Python 3.12 이상 표준 라이브러리만 사용한다. 실행 중 읽고 쓰는 파일은 모두 TemporaryDirectory가 만든 폴더 안에 있다. 가짜 모델은 지원하는 목표와 수정 요청에만 정해진 계획을 돌려준다. 검사를 통과하지 못하면 예외로 종료하며, 정상 실행은 아래 실행 결과와 같은 출력을 낸다.

import json
import tempfile
from pathlib import Path

GOAL = "메모를 정리해 번호 목록 만들기"
REVISION = "목록 전에 항목 수 확인 추가"
RAW = ["  준비물 확인  ", "", "회의 메모 ", " 준비물 확인"]


def require(condition, message):
    if not condition:
        raise ValueError(message)


def read_json(path):
    return json.loads(path.read_text(encoding="utf-8"))


def write_json(path, value):
    temporary = path.with_name(path.name + ".tmp")
    temporary.write_text(
        json.dumps(value, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    temporary.replace(path)


def task(task_id):
    return {"id": task_id, "tool": task_id, "check": task_id}


def fake_model(request, current=None):
    if request == GOAL and current is None:
        return {
            "version": 1,
            "tasks": [task("collect"), task("normalize"), task("report")],
        }
    if request == REVISION and current is not None:
        tasks = []
        for item in current["tasks"]:
            if item["id"] == "report":
                tasks.append(task("audit"))
            tasks.append(dict(item))
        return {"version": current["version"] + 1, "tasks": tasks}
    raise ValueError("지원하지 않는 요청")


def clean_items(items):
    return [item.strip() for item in items if item.strip()]


def report_text(items):
    return "".join(
        f"{number}. {item}\n"
        for number, item in enumerate(items, start=1)
    )


def collect(root):
    write_json(root / "raw.json", RAW)


def normalize(root):
    write_json(
        root / "clean.json",
        clean_items(read_json(root / "raw.json")),
    )


def audit(root):
    write_json(
        root / "audit.json",
        {"count": len(read_json(root / "clean.json"))},
    )


def report(root):
    (root / "report.txt").write_text(
        report_text(read_json(root / "clean.json")),
        encoding="utf-8",
    )


def check_collect(root):
    return read_json(root / "raw.json") == RAW


def check_normalize(root):
    return (
        read_json(root / "clean.json")
        == clean_items(read_json(root / "raw.json"))
    )


def check_audit(root):
    return read_json(root / "audit.json") == {
        "count": len(read_json(root / "clean.json"))
    }


def check_report(root):
    return (
        (root / "report.txt").read_text(encoding="utf-8")
        == report_text(read_json(root / "clean.json"))
    )


TOOLS = {
    "collect": collect,
    "normalize": normalize,
    "audit": audit,
    "report": report,
}
CHECKS = {
    "collect": check_collect,
    "normalize": check_normalize,
    "audit": check_audit,
    "report": check_report,
}


def validate_plan(plan):
    require(type(plan["version"]) is int, "계획 버전은 정수여야 한다")
    require(plan["version"] > 0, "계획 버전은 양수여야 한다")
    tasks = plan["tasks"]
    require(isinstance(tasks, list) and tasks, "작업 목록이 필요하다")
    ids = []
    for item in tasks:
        require(
            set(item) == {"id", "tool", "check"},
            "작업 필드가 올바르지 않다",
        )
        require(isinstance(item["id"], str), "식별자는 문자열이어야 한다")
        require(item["id"] in TOOLS, "알 수 없는 작업")
        require(item["tool"] == item["id"], "작업과 도구가 맞지 않는다")
        require(item["check"] == item["id"], "작업과 확인 방법이 맞지 않는다")
        ids.append(item["id"])
    require(len(ids) == len(set(ids)), "식별자가 중복되었다")
    require(
        ids in [
            ["collect", "normalize", "report"],
            ["collect", "normalize", "audit", "report"],
        ],
        "지원하지 않는 작업 순서",
    )


def prepare_state(root, plan):
    path = root / "state.json"
    old = read_json(path) if path.exists() else None
    if old is not None:
        require(
            plan["version"] >= old["plan_version"],
            "계획 버전이 이전으로 돌아갔다",
        )
        previous = old["tasks"]
        definitions = [entry["definition"] for entry in previous.values()]
        if plan["version"] == old["plan_version"]:
            require(definitions == plan["tasks"], "같은 버전의 계획이 바뀌었다")

        completed = []
        pending_seen = False
        for task_id, entry in previous.items():
            status = entry["status"]
            require(status in {"pending", "running", "done"}, "알 수 없는 상태")
            if status == "done":
                require(not pending_seen, "완료 작업은 앞부분에 모여야 한다")
                completed.append(task_id)
            else:
                pending_seen = True

        new_ids = [item["id"] for item in plan["tasks"]]
        require(
            new_ids[:len(completed)] == completed,
            "완료 작업의 순서가 바뀌었다",
        )

    state = {"plan_version": plan["version"], "tasks": {}}
    for item in plan["tasks"]:
        task_id = item["id"]
        entry = old["tasks"].get(task_id) if old is not None else None
        status = "pending"
        if entry is not None and entry["status"] == "done":
            require(entry["definition"] == item, "완료 작업의 정의가 바뀌었다")
            require(CHECKS[item["check"]](root), "완료 산출물 확인 실패")
            status = "done"
        elif entry is not None and entry["status"] == "running":
            print(f"재시도 준비: {task_id}")
        state["tasks"][task_id] = {
            "definition": dict(item),
            "status": status,
        }
    write_json(path, state)
    return state


def run(root, stop_after=None):
    plan = read_json(root / "plan.json")
    validate_plan(plan)
    state = prepare_state(root, plan)
    executed = 0
    for item in plan["tasks"]:
        task_id = item["id"]
        entry = state["tasks"][task_id]
        if entry["status"] == "done":
            print(f"건너뜀: {task_id} (완료 확인)")
            continue
        entry["status"] = "running"
        write_json(root / "state.json", state)
        TOOLS[item["tool"]](root)
        require(CHECKS[item["check"]](root), f"완료 확인 실패: {task_id}")
        entry["status"] = "done"
        write_json(root / "state.json", state)
        print(f"완료: {task_id}")
        executed += 1
        if stop_after is not None and executed == stop_after:
            print(f"중단: 이번 실행에서 {executed}개 완료")
            return False
    print("전체 작업 완료")
    return True


def main():
    with tempfile.TemporaryDirectory() as folder:
        root = Path(folder)
        initial = fake_model(GOAL)
        validate_plan(initial)
        write_json(root / "plan.json", initial)

        print("첫 실행")
        require(run(root, stop_after=2) is False, "중단 확인 실패")
        paused = read_json(root / "state.json")
        require(
            [entry["status"] for entry in paused["tasks"].values()]
            == ["done", "done", "pending"],
            "중단 상태 확인 실패",
        )

        revised = fake_model(REVISION, read_json(root / "plan.json"))
        validate_plan(revised)
        require(
            [item for item in revised["tasks"] if item["id"] != "audit"]
            == initial["tasks"],
            "기존 작업 정의가 바뀌었다",
        )
        write_json(root / "plan.json", revised)
        print("계획 수정: 버전 1 → 2")
        print("변경 전: collect → normalize → report")
        print("변경 후: collect → normalize → audit → report")

        print("재시작")
        require(run(root) is True, "완료 확인 실패")
        final_state = read_json(root / "state.json")
        require(final_state["plan_version"] == 2, "상태 버전 확인 실패")
        require(
            all(entry["status"] == "done"
                for entry in final_state["tasks"].values()),
            "최종 상태 확인 실패",
        )
        require(read_json(root / "audit.json") == {"count": 3}, "수량 확인 실패")
        print("최종 상태: 모든 작업 done")
        print("항목 수: 3")
        print("최종 문서")
        print((root / "report.txt").read_text(encoding="utf-8"), end="")


if __name__ == "__main__":
    main()

줄별 해설

가져오기와 상수. json은 계획과 상태를 직렬화하고, tempfile은 예제 파일의 범위를 임시 폴더로 제한한다. Path는 파일 경로를 다룬다. GOAL과 REVISION은 가짜 모델이 구별하는 요청이다. RAW에는 공백, 빈 항목, 중복 항목을 넣어 정리 규칙을 확인한다. 외부 입력이나 실행 시각에 따라 달라지는 값은 없다.

require, read_json, write_json. require는 조건이 거짓이면 예외를 낸다. 실행 옵션에 따라 생략될 수 있는 assert 대신 일반 조건문으로 검사를 수행한다. read_json은 UTF-8로 읽은 문자열을 파이썬 값으로 바꾼다. write_json은 같은 폴더의 임시 파일에 내용을 완성하고 replace로 대상 파일을 교체한다. 한글은 읽기 쉽게 그대로 저장한다.

task와 fake_model. task는 한 작업의 세 필드를 만든다. 이번 예제에서는 작업 식별자, 도구 이름, 확인 이름을 같게 하여 대응 관계를 단순하게 유지한다. fake_model은 최초 요청에 세 작업을 반환한다. 수정 요청에서는 report를 만나기 직전에 audit를 넣는다. 기존 작업은 dict로 복사하므로 새 목록을 만들면서 이전 작업 딕셔너리를 직접 바꾸지 않는다.

clean_items와 report_text. 두 함수는 파일을 건드리지 않고 값만 변환한다. clean_items는 공백을 제거한 결과가 비어 있지 않은 항목을 남긴다. report_text는 1부터 시작하는 번호와 줄바꿈을 붙인다. 이런 작은 계산 함수를 따로 두면 파일을 읽는 부분과 결과를 만드는 부분을 구분해서 확인하기 쉽다.

네 도구와 네 확인 함수. 도구는 각각 자기 산출물을 저장한다. 확인 함수는 파일을 다시 읽어 기대하는 값과 비교한다. 예를 들어 check_normalize는 clean.json이 존재하는지만 검사하지 않고 raw.json에서 계산한 배열과 비교한다. check_report는 번호, 순서, 내용, 마지막 줄바꿈까지 확인한다. 파일을 읽지 못하면 예외가 발생하므로 완료 처리가 진행되지 않는다.

TOOLS와 CHECKS. 두 딕셔너리는 계획에 있는 문자열을 실제 함수와 연결한다. 계획 문자열을 파이썬 코드로 평가하지 않는다. 허용된 이름을 가진 함수만 실행한다. 이번 장에서는 계획과 상태에 집중하기 위해 도구의 입력 경로도 코드 안에 고정한다.

validate_plan. 버전, 작업 필드, 식별자 중복, 도구와 확인 방법의 대응, 작업 순서를 검사한다. 정수 검사에서 type을 쓰는 이유는 파이썬에서 bool도 int의 하위 유형이기 때문이다. 이 예제의 계획은 두 가지 순서만 허용한다. 일반적인 계획 언어를 구현한 것이 아니라 현재 목표에 필요한 입력 계약을 좁게 정한 것이다.

prepare_state의 앞부분. 상태 파일이 있으면 계획 버전이 뒤로 돌아가지 않았는지 확인한다. 같은 버전인데 작업 정의가 바뀌었다면 거부한다. 상태 파일의 작업 순서는 이 프로그램이 계획 순서대로 저장한다. 그 순서에서 done이 연속된 앞부분에 있는지 검사하고, 새 계획도 같은 완료 작업들로 시작하는지 확인한다.

prepare_state의 뒷부분. 새 계획에 맞춰 상태를 다시 구성한다. 완료 작업은 정의가 같고 산출물도 확인될 때만 done을 유지한다. 실행 중이던 작업과 미완료 작업은 pending으로 둔다. 추가된 audit도 이전 상태에 없으므로 pending이 된다. 마지막에는 연결된 계획 버전을 함께 저장한다.

run. 호출마다 계획과 상태를 파일에서 읽는다. done은 이미 prepare_state에서 확인했으므로 실행을 건너뛴다. 나머지는 running 저장, 도구 실행, 결과 확인, done 저장 순서로 처리한다. stop_after는 이번 호출에서 새로 완료한 작업 수만 센다. 이전 호출에서 끝난 작업은 중단 횟수에 포함되지 않는다.

main. 첫 실행을 두 작업 뒤에 멈추고, 파일에 남은 상태가 예상과 같은지 확인한다. 이후 새 계획에서 audit를 제외한 목록이 원래 계획과 같은지 비교한다. 재시작 뒤에는 버전, 모든 작업의 완료 여부, 항목 수를 검사한다. 이것이 예제 안에 포함된 실행 확인이다. 확인 함수와 도구가 계산 함수를 공유하므로, 계산 규칙 자체는 연습 문제처럼 구체적인 입력과 기대값으로 별도 점검해야 한다.

실행 결과

실행 명령은 다음과 같다. 별도의 패키지 설치나 환경 변수 설정은 필요하지 않다.

python3 main.py

예상 출력은 다음과 같다. 첫 실행에는 두 작업만 완료된다. 재시작에서는 기존 두 작업의 산출물을 확인한 뒤 건너뛰고, 추가한 audit와 남아 있던 report를 실행한다.

첫 실행
완료: collect
완료: normalize
중단: 이번 실행에서 2개 완료
계획 수정: 버전 1 → 2
변경 전: collect → normalize → report
변경 후: collect → normalize → audit → report
재시작
건너뜀: collect (완료 확인)
건너뜀: normalize (완료 확인)
완료: audit
완료: report
전체 작업 완료
최종 상태: 모든 작업 done
항목 수: 3
최종 문서
1. 준비물 확인
2. 회의 메모
3. 준비물 확인

중복 항목이 남아 있는 것은 명세대로다. 항목 수는 빈 문자열을 제외한 뒤의 세 개이며, 고유한 문자열 개수와 다르다. 출력이 원하는 모습이라는 인상만으로 확인을 끝내지 말고, 공백 처리와 중복 유지가 요구와 일치하는지 비교한다. 코드를 수정했다면 경고를 오류로 취급하는 컴파일 확인도 실행할 수 있다. 다음 명령은 컴파일 캐시 역시 임시 폴더에 두고 성공하면 한 줄을 출력한다.

python3 -W error -c 'import py_compile, tempfile; from pathlib import Path; folder = tempfile.TemporaryDirectory(); py_compile.compile("main.py", cfile=str(Path(folder.name) / "main.pyc"), doraise=True); folder.cleanup(); print("컴파일 확인 완료")'

이 명령의 예상 출력은 다음과 같다. 컴파일 확인은 문법과 컴파일 단계의 경고를 확인하며, 실행 결과 검사를 대신하지 않는다.

컴파일 확인 완료

실무에서 자주 틀리는 것

도구를 부르기 전에 완료로 표시한다

아래 코드는 결과가 확인되기 전에 done으로 바꾼다. 그 시점에 멈추면 실행하지 않은 작업이 완료로 남는다. 출력문은 실행 위치를 보여 주는 간단한 대역이다.

entry = {"status": "pending"}
entry["status"] = "done"
print("도구 실행 위치")

완료는 실행과 결과 확인을 마친 뒤에 기록한다. 아래 순서를 실제 하네스에 적용할 때는 running과 done을 바꾸는 지점마다 상태 파일을 저장한다.

entry = {"status": "pending"}
entry["status"] = "running"
print("도구 실행 위치")
verified = True
if not verified:
    raise ValueError("완료 확인 실패")
entry["status"] = "done"

작업 위치를 상태의 식별자로 쓴다

다음 코드는 두 번째 작업을 완료한 뒤 그 앞에 새 작업을 삽입한다. 상태의 두 번째 값은 여전히 done이어서 audit가 완료된 것처럼 연결된다.

tasks = ["collect", "normalize", "report"]
statuses = ["done", "done", "pending"]
tasks.insert(1, "audit")
print(tasks[1], statuses[1])

상태를 식별자에 연결하면 위치가 달라져도 주인이 바뀌지 않는다. 다만 완성 코드에서는 완료 작업 앞에 삽입하는 변경 자체도 거부한다. 식별자 연결과 변경 정책은 함께 필요하다.

tasks = ["collect", "normalize", "report"]
statuses = {"collect": "done", "normalize": "done", "report": "pending"}
tasks.insert(1, "audit")
statuses["audit"] = "pending"
print(tasks[1], statuses[tasks[1]])

재시작할 때 상태를 새로 만든다

상태를 파일에 썼더라도 시작할 때 늘 pending으로 덮어쓰면 이어서 처리할 근거가 사라진다. 다음 예제는 임시 폴더 안의 저장된 done을 즉시 잃어버린다.

import json
import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory() as folder:
    path = Path(folder) / "state.json"
    path.write_text('{"collect": "done"}', encoding="utf-8")
    state = {"collect": "pending"}
    path.write_text(json.dumps(state), encoding="utf-8")
    print(state["collect"])

기존 파일이 있으면 먼저 읽는다. 그다음 계획 정의와 산출물을 확인해야 한다. 다음 코드는 읽기와 초기화의 구분만 보여 주며, 완성 코드의 prepare_state가 추가 확인을 담당한다.

import json
import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory() as folder:
    path = Path(folder) / "state.json"
    path.write_text('{"collect": "done"}', encoding="utf-8")
    if path.exists():
        state = json.loads(path.read_text(encoding="utf-8"))
    else:
        state = {"collect": "pending"}
    print(state["collect"])

같은 식별자의 완료 작업을 다른 작업으로 바꾼다

식별자만 비교하면 정리 규칙이 달라져도 이전 완료 상태를 가져올 수 있다. 다음 예제에서는 같은 normalize에 다른 도구가 붙었는데도 완료를 유지한다.

old = {"id": "normalize", "tool": "trim", "check": "trim"}
new = {"id": "normalize", "tool": "deduplicate", "check": "deduplicate"}
status = "done" if old["id"] == new["id"] else "pending"
print(status)

완료 작업은 정의까지 같아야 상태를 유지한다. 아래 예제에서는 pending이 출력된다. 완성 코드는 완료 정의의 변경을 발견하면 자동 초기화 대신 예외를 내므로, 그 결과에 의존하는 후속 작업을 함께 검토할 수 있다.

old = {"id": "normalize", "tool": "trim", "check": "trim"}
new = {"id": "normalize", "tool": "deduplicate", "check": "deduplicate"}
status = "done" if old == new else "pending"
print(status)

한눈에 보기

중단과 계획 수정에 견디는 순차 실행의 판단 기준

상황	판단 기준	예제의 처리
작업을 나눌 때	입력, 산출물, 확인 방법이 분명한가	작업에 도구와 확인 이름을 붙인다
도구가 반환했을 때	실제 산출물이 요구를 만족하는가	파일을 다시 읽고 비교한다
done을 다시 읽을 때	정의와 산출물이 여전히 맞는가	확인한 뒤 실행을 건너뛴다
running을 다시 읽을 때	같은 작업을 재실행해도 되는가	덮어쓰기 도구를 pending으로 돌린다
계획을 수정할 때	완료 작업의 의미와 순서를 보존하는가	버전을 올리고 정의를 비교한다
확인이 실패했을 때	완료로 기록할 근거가 있는가	예외를 내고 멈춘다

계획 파일이 길다고 실행이 안정되는 것은 아니다. 작업마다 확인 가능한 결과가 있고, 상태를 저장하는 순서가 분명해야 한다. 다음 장에서는 실행에 필요한 정보를 어떻게 고르고 압축해 전달할지 다룬다. 작업 목록과 상태 파일은 그때도 전체 자료를 다시 읽는 대신 현재 필요한 작업을 고르는 출발점이 된다.

연습 문제

1. 첫 실행의 stop_after를 1로 바꾼다. 중단 상태 검사도 맞춰 수정한다. 재시작에서 건너뛰는 작업과 새로 실행하는 작업을 각각 예상하고 실제 출력과 비교한다.
2. clean_items를 호출해 공백만 있는 항목, 중복 항목, 양끝 공백이 있는 항목을 확인하는 검사를 추가한다. 기대값은 함수 호출로 만들지 말고 직접 적는다.
3. 첫 실행 직후 normalize의 상태를 running으로 바꾸어 저장한다. 재시작이 normalize를 다시 실행하는지 확인하고, 최종 항목 수가 늘어나지 않는 이유를 설명한다.
4. 재시작 전에 clean.json의 내용을 빈 배열로 덮어쓴다. 어느 검사에서 실행이 멈추는지 확인한다. 이 상황에서 normalize만 pending으로 바꾸는 정책을 일반화하려면 무엇을 추가로 판단해야 하는지 설명한다.

정답과 해설

1. 첫 실행은 collect만 완료한다. 중단 상태의 기대값은 done, pending, pending이다. 재시작은 collect를 건너뛰고 normalize, audit, report를 차례대로 실행한다. 계획 수정은 미완료 작업 사이에 audit를 넣으므로 완료한 collect의 위치와 정의를 바꾸지 않는다. 최종 문서와 항목 수는 원래 예제와 같다.

2. 아래 함수는 입력과 기대값을 직접 지정한다. 완성 코드의 clean_items 정의 뒤에 추가하고 main에서 호출한다. 중복을 제거하는 수정이나 공백 제거를 빠뜨리는 수정이 들어오면 실패한다.

   def test_clean_items():
       cases = [
           (["   ", ""], []),
           (["가", "가"], ["가", "가"]),
           (["  메모  ", "기록 "], ["메모", "기록"]),
       ]
       for source, expected in cases:
           require(clean_items(source) == expected, "정리 규칙 확인 실패")
   

3. main의 중단 상태 검사가 끝난 뒤 아래 코드를 넣는다. 재시작에서는 collect만 건너뛰고 normalize의 재시도 준비 메시지를 출력한다. normalize는 원본 배열에서 다시 계산한 결과를 clean.json에 덮어쓰므로 항목이 누적되지 않는다. 이미 만들어진 파일이 있어도 running 상태만으로는 완료를 인정하지 않는다.

   paused["tasks"]["normalize"]["status"] = "running"
   write_json(root / "state.json", paused)
   

4. 재시작 호출 전에 아래 코드를 넣으면 prepare_state에서 normalize의 완료 산출물 확인이 실패한다. 빈 배열은 원본에서 계산한 세 항목과 다르기 때문이다. 예외가 발생하므로 audit와 report는 실행되지 않는다.

   write_json(root / "clean.json", [])
   

   자동 복구로 확장하려면 바뀐 산출물에 의존하는 후속 작업의 상태도 판단해야 한다. 후속 결과가 이미 만들어졌다면 함께 무효화할 필요가 있다. 원본이 그대로인지, 재실행이 안전한지, 현재 계획이 동일한 정리 규칙을 요구하는지도 확인해야 한다. 이번 예제는 자동 복구를 추측해서 진행하지 않고 불일치를 드러내는 정책을 사용한다.