diff --git a/harness/nsp/taskdoc.py b/harness/nsp/taskdoc.py
new file mode 100644
index 00000000..25182f39
--- /dev/null
+++ b/harness/nsp/taskdoc.py
@@ -0,0 +1,69 @@
+"""Per-step task document generator (spec D16, §4.9).
+
+Emette un singolo documento compatto per fase/step, derivato dagli artefatti precedenti
+e dalla vista effective delle decisioni, con byte budget enforced (target <20k token
+per un modello 35B/<200k). MAI incorpora physical.yaml (~190k token, fatale).
+
+D15+D16 complementari: il task doc e' generato dalla vista effective_decisions, quindi
+post-rollback riflette automaticamente lo stato corretto (le decisioni stale di fasi
+> current_phase sono escluse).
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from nsp.phase import effective_decisions
+from nsp.workflow import load_workflow
+
+MAX_BODY_BYTES = 80_000  # ~20k token (target per task document di una fase)
+
+
+@dataclass
+class TaskDoc:
+    phase: int
+    body: str
+    byte_budget_ok: bool
+
+
+def generate_task_doc(
+    session_dir: Path | str,
+    phase: int,
+    promoted_tables: list[str] | None = None,
+) -> TaskDoc:
+    """Genera il documento di task per la fase `phase`.
+
+    Contenuto (compatti, mai artefatti integrali fatali):
+    - Domanda (question.md) se presente.
+    - Schema linking (schema_linking.json) solo da fase >= 4.
+    - Brief delle decisioni effective (esclude stale post-rollback, esclude ritirate).
+    - Header del task con il numero/nome della fase.
+    """
+    session_dir = Path(session_dir)
+    parts: list[str] = []
+
+    q = session_dir / "question.md"
+    if q.exists():
+        parts.append("## Domanda\n" + q.read_text())
+
+    sl = session_dir / "schema_linking.json"
+    if sl.exists() and phase >= 4:
+        parts.append("## Schema linking (deciso)\n```json\n" + sl.read_text() + "\n```")
+
+    # Brief decisioni effective (D15-aware)
+    eff = effective_decisions(session_dir)
+    if eff:
+        lines = [f"- {d.type} | {d.subject} | {d.detail}" for d in eff]
+        parts.append("## Decisioni effettive (effective)\n" + "\n".join(lines))
+
+    # Header fase
+    try:
+        wf = load_workflow()
+        name = wf.phase_name(phase)
+        header = f"## Task: fase {phase} ({name})"
+    except Exception:
+        header = f"## Task: fase {phase}"
+    parts.append(header)
+
+    body = "\n\n".join(parts)
+    return TaskDoc(phase=phase, body=body, byte_budget_ok=len(body.encode()) <= MAX_BODY_BYTES)
diff --git a/harness/tests/test_taskdoc.py b/harness/tests/test_taskdoc.py
new file mode 100644
index 00000000..977216e1
--- /dev/null
+++ b/harness/tests/test_taskdoc.py
@@ -0,0 +1,70 @@
+from pathlib import Path
+
+from nsp.taskdoc import generate_task_doc
+
+
+def test_task_doc_includes_question_and_schema_scope(tmp_path):
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "question.md").write_text("# Domanda\nQuanti pazienti?\n## Assunzioni\n- a")
+    (s / "schema_linking.json").write_text(
+        '{"question":"q","candidates":[{"kind":"table","name":"pazienti"}],"joins":[],"excluded":[],"open_questions":[]}'
+    )
+    doc = generate_task_doc(session_dir=s, phase=7)
+    assert "Quanti pazienti?" in doc.body
+    assert "pazienti" in doc.body
+    assert doc.byte_budget_ok is True
+
+
+def test_task_doc_never_embeds_full_physical_yaml(tmp_path):
+    """physical.yaml e' fatale per un 35B/<200k (~190k token). Mai incorporarlo."""
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "question.md").write_text("q")
+    # un physical.yaml enorme fuori dalla sessione (come in ChironeWp3: artifacts/mschema/)
+    (s.parent / "physical.yaml").write_text("x: " + "y" * 800_000)
+    doc = generate_task_doc(session_dir=s, phase=7)
+    assert "physical.yaml" not in doc.body
+    assert len(doc.body) < 100_000  # bounded
+
+
+def test_task_doc_byte_budget_enforced_on_normal_input(tmp_path):
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "question.md").write_text("q")
+    doc = generate_task_doc(session_dir=s, phase=1)
+    assert doc.byte_budget_ok is True
+
+
+def test_task_doc_byte_budget_violation_flagged(tmp_path):
+    """Se un artefatto di sessione e' enorme (input perverso), byte_budget_ok diventa False."""
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "question.md").write_text("q")
+    (s / "schema_linking.json").write_text("x: " + "y" * 400_000)  # ~400KB -> over budget
+    doc = generate_task_doc(session_dir=s, phase=7)
+    assert doc.byte_budget_ok is False
+
+
+def test_task_doc_carries_phase_header(tmp_path):
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "question.md").write_text("q")
+    doc = generate_task_doc(session_dir=s, phase=4)
+    assert "fase 4" in doc.body.lower() or "fase 4" in doc.body
+
+
+def test_task_doc_excludes_stale_decisions_post_rollback(tmp_path):
+    """D15+D16: il task doc riflette lo stato effective, non quello stale.
+    Dopo rollback a F4, una sql_approved:7 stale non appare nel brief delle decisioni."""
+    from nsp.decisions import append_decision
+    from nsp.phase import current_phase
+    s = tmp_path / "sess"; s.mkdir()
+    (s / "question.md").write_text("q")
+    # simula: lavoro fino a F7, poi rollback a F4
+    append_decision(s, type="phase_approved", subject="phase:1")
+    append_decision(s, type="phase_approved", subject="phase:2")
+    append_decision(s, type="phase_approved", subject="phase:3")
+    append_decision(s, type="phase_approved", subject="phase:4")
+    append_decision(s, type="sql_approved", subject="phase:7", detail="SELECT 1")
+    append_decision(s, type="phase_reopened", subject="phase:4")
+    assert current_phase(s) == 4
+    doc = generate_task_doc(session_dir=s, phase=4)
+    # la decisione stale di fase 7 NON deve apparire nel brief
+    assert "sql_approved" not in doc.body
+    assert "phase:7" not in doc.body