diff --git a/core/lambda/nova_idp_token_vend.py b/core/lambda/nova_idp_token_vend.py
new file mode 100644
index 0000000..ed8ddb3
--- /dev/null
+++ b/core/lambda/nova_idp_token_vend.py
@@ -0,0 +1,391 @@
+"""Nova IdP token-vend Lambda — PAT/session → KMS-signed OIDC token
+(REQ-336, C-6.1/C-7.1 ABAC FAIL-CLOSED, D-229 revocation).
+
+Accepts a PAT (or session token) and returns a KMS-signed OIDC token
+with claims ``sub, aud, iss, exp, iat, jti, roles`` (REQ-336).
+
+## ABAC fail-closed (C-6.1/C-7.1 — INV-17 runtime enforcement)
+
+The grill's #1 finding: the token-vend Lambda MUST fail closed on ABAC
+evaluation failure. Concretely, a token is vended **only** when:
+
+1. The PAT is active (``nova-pats.GetItem(jti, ConsistentRead=True)``
+   returns an item with ``status == "active"`` — D-229; strong read on
+   the main table, GSIs don't support strong reads).
+2. ``KyvernoJsonEngine.is_configured()`` returns ``True`` **AND**
+   ``evaluate_token_vend_policy()`` returns ``allowed=True`` without
+   raising.
+
+If (2) fails for **any** reason — ``kj`` absent, ``kj`` error, policy
+parse error, engine raise — the Lambda returns **403** + audit
+``token.vend.denied`` (reason ``abac_eval_failed``). **Never fail
+open.** This is verified by ``tests/test_abac_fail_closed.py`` — the
+most important test of the milestone.
+
+## Dual-use (REQ-329 pattern)
+
+Mirrors ``nova_idp_auth.py``: lazy boto3, env-var table names,
+``NOVA_LAMBDA_LOCAL_BYPASS``, ``__main__`` CLI block, audit emission.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import os
+import sys
+import time
+
+import boto3
+
+# ---------------------------------------------------------------------------
+# Config (env-var table names, mirroring nova_idp_auth.py)
+# ---------------------------------------------------------------------------
+
+PATS_TABLE = os.environ.get("NOVA_PATS_TABLE", "nova-pats")
+SESSIONS_TABLE = os.environ.get("NOVA_SESSIONS_TABLE", "nova-sessions")
+OIDC_KMS_KEY_ID = os.environ.get("NOVA_OIDC_KMS_KEY_ID", "alias/nova-oidc-signing")
+OIDC_ISSUER = os.environ.get("NOVA_OIDC_ISSUER", "nova-idp")
+OIDC_AUDIENCE = os.environ.get("NOVA_OIDC_AUDIENCE", "nova-cli")
+# OIDC token lifetime (seconds). Default 15 min.
+OIDC_TTL_SECONDS = int(os.environ.get("NOVA_OIDC_TTL_SECONDS", "900"))
+
+_dynamodb = None
+_kms_client = None
+
+
+def _get_dynamodb():
+    """Lazy boto3 DynamoDB resource singleton (mirrors contract_ingestor)."""
+    global _dynamodb
+    if _dynamodb is None:
+        _dynamodb = boto3.resource("dynamodb")
+    return _dynamodb
+
+
+def _iso8601_now() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).strftime(
+        "%Y-%m-%dT%H:%M:%SZ"
+    )
+
+
+def _epoch_now() -> int:
+    return int(datetime.datetime.now(datetime.timezone.utc).timestamp())
+
+
+def _emit_audit(event_type: str, **fields) -> None:
+    """Emit an audit event to stderr as JSON (never the raw PAT/token)."""
+    payload = {"event": event_type, "ts": _iso8601_now(), **fields}
+    # Defense-in-depth: scrub raw token fields (INV-16/INV-17 spirit).
+    for _k in ("pat", "session_token", "token", "raw_pat"):
+        payload.pop(_k, None)
+    sys.stderr.write(json.dumps(payload, sort_keys=True) + "\n")
+    sys.stderr.flush()
+
+
+# ---------------------------------------------------------------------------
+# PAT / session decoding (decode WITHOUT verifying — signature verified
+# by KMS public key separately at the JWKS verifier; the revocation
+# check is the trust anchor here, not the JWT signature).
+# ---------------------------------------------------------------------------
+
+
+def _decode_jwt_unverified(token: str) -> dict:
+    """Decode a JWT's payload without verifying the signature."""
+    try:
+        import jwt as pyjwt
+        return pyjwt.decode(token, options={"verify_signature": False})
+    except Exception:
+        # Fallback: manual base64url decode of the payload segment.
+        parts = token.split(".")
+        if len(parts) < 2:
+            raise ValueError("malformed JWT (expected 3 segments)")
+        import base64
+        pad = parts[1] + "=" * (-len(parts[1]) % 4)
+        return json.loads(base64.urlsafe_b64decode(pad))
+
+
+def _extract_pat_claims(token: str) -> dict:
+    """Decode a PAT/session JWT → extract jti, sub, typ, roles, owner, exp."""
+    claims = _decode_jwt_unverified(token)
+    required = ("jti", "sub", "exp")
+    for f in required:
+        if f not in claims:
+            raise ValueError(f"token missing claim: {f}")
+    return claims
+
+
+# ---------------------------------------------------------------------------
+# Revocation check (D-229 — strong read on the main table)
+# ---------------------------------------------------------------------------
+
+
+def _check_pat_active(jti: str) -> tuple[bool, str]:
+    """Return ``(active, reason)``. Strong read on nova-pats main table.
+
+    D-229: GSIs don't support strongly-consistent reads, so the
+    revocation check uses ``GetItem(PK=jti, ConsistentRead=True)`` on
+    the main table. This satisfies the 60s SLO synchronously (the
+    strong read reflects the latest write — revocation is instant).
+    """
+    table = _get_dynamodb().Table(PATS_TABLE)
+    resp = table.get_item(
+        TableName=PATS_TABLE,
+        Key={"jti": jti},
+        ConsistentRead=True,
+    )
+    item = resp.get("Item")
+    if item is None:
+        return False, "pat_unknown"
+    status = item.get("status", "active")
+    if status != "active":
+        return False, f"pat_{status}"  # pat_revoked, pat_expired, etc.
+    # Expired? (defense-in-depth; TTL may not have reaped it yet)
+    expires_at = item.get("expires_at")
+    if expires_at is not None:
+        try:
+            if int(expires_at) < _epoch_now():
+                return False, "pat_expired"
+        except (ValueError, TypeError):
+            pass
+    return True, "active"
+
+
+# ---------------------------------------------------------------------------
+# ABAC fail-closed (C-6.1/C-7.1)
+# ---------------------------------------------------------------------------
+
+
+def _build_abac_payload(claims: dict, requested_claims: list[str],
+                        target_resource: dict, environment: str,
+                        policy_version: str) -> dict:
+    """Build the ABAC authorization payload (REQ-339, C-5.1)."""
+    return {
+        "subject": {
+            "id": claims.get("sub", ""),
+            "role": (claims.get("roles") or ["unknown"])[0],
+            "owner": claims.get("owner", ""),
+        },
+        "requested_claims": requested_claims,
+        "target_resource": target_resource,
+        "environment": environment,
+        "pat_jti": claims.get("jti", ""),
+        "policy_version": policy_version,
+    }
+
+
+def _evaluate_abac_fail_closed(payload: dict) -> tuple[bool, list, str, str]:
+    """Evaluate ABAC with fail-closed semantics (C-6.1).
+
+    Returns ``(allowed, pcrs, policy_sha, reason)``. On ANY failure
+    (engine not configured, evaluate raises, policy parse error) returns
+    ``(False, [], "", "abac_eval_failed")``. **Never fails open.**
+    """
+    # Lazy imports so the module imports without the engine adapter.
+    from core.policy_engine import get_engine
+
+    # C-6.1: is_configured() check. If kj is absent → fail closed.
+    try:
+        engine = get_engine()
+        if not engine.is_configured():
+            _emit_audit(
+                "token.vend.abac_engine_not_configured",
+                pat_jti=payload.get("pat_jti", ""),
+            )
+            return False, [], "", "abac_eval_failed"
+    except Exception:  # noqa: BLE001 - fail closed on any engine check error
+        return False, [], "", "abac_eval_failed"
+
+    # C-6.1: evaluate() raising → fail closed.
+    try:
+        from core.abac_evaluator import evaluate_token_vend_policy
+        allowed, pcrs, policy_sha = evaluate_token_vend_policy(payload)
+        reason = "abac_denied" if not allowed else "ok"
+        return allowed, pcrs, policy_sha, reason
+    except Exception:  # noqa: BLE001 - fail closed on any eval error
+        return False, [], "", "abac_eval_failed"
+
+
+# ---------------------------------------------------------------------------
+# Token vend (REQ-336)
+# ---------------------------------------------------------------------------
+
+
+def _build_oidc_claims(pat_claims: dict) -> dict:
+    """Build the OIDC token claims (REQ-336)."""
+    now = _epoch_now()
+    return {
+        "sub": pat_claims["sub"],
+        "aud": OIDC_AUDIENCE,
+        "iss": OIDC_ISSUER,
+        "exp": now + OIDC_TTL_SECONDS,
+        "iat": now,
+        "jti": pat_claims.get("jti", ""),  # carry the PAT jti for tracing
+        "roles": pat_claims.get("roles", []),
+        "typ": "nova_oidc_token",  # INV-14: distinguish from developer_pat
+    }
+
+
+def vend_token(
+    token: str,
+    requested_claims: list[str] | None = None,
+    target_resource: dict | None = None,
+    environment: str | None = None,
+    policy_version: str = "",
+) -> dict:
+    """Vend a KMS-signed OIDC token for a PAT/session (REQ-336).
+
+    Returns ``{"token": ..., "expires_at": ...}`` on success. Raises
+    ``_DeniedError`` (→ 403) on revocation / ABAC denial.
+    """
+    requested_claims = requested_claims or ["sub", "roles"]
+    target_resource = target_resource or {"type": "contract", "id": "*", "owner": "*", "environment": environment or "dev"}
+    environment = environment or "dev"
+
+    # 1. Decode the PAT/session (without verifying — D-229).
+    pat_claims = _extract_pat_claims(token)
+    jti = pat_claims["jti"]
+
+    # 2. Revocation check (D-229, strong read).
+    active, reason = _check_pat_active(jti)
+    if not active:
+        _emit_audit("token.vend.denied", pat_jti=jti, reason=reason)
+        raise _DeniedError(reason)
+
+    # 3. ABAC eval (C-6.1 FAIL-CLOSED).
+    abac_payload = _build_abac_payload(
+        pat_claims, requested_claims, target_resource, environment, policy_version
+    )
+    allowed, _pcrs, policy_sha, abac_reason = _evaluate_abac_fail_closed(abac_payload)
+    if not allowed:
+        _emit_audit(
+            "token.vend.denied",
+            pat_jti=jti,
+            reason=abac_reason,
+            policy_sha=policy_sha,
+        )
+        raise _DeniedError(abac_reason)
+
+    # 4. KMS sign (REQ-337).
+    from core.kms_signing import sign_jwt
+    oidc_claims = _build_oidc_claims(pat_claims)
+    oidc_token = sign_jwt(oidc_claims, key_id=OIDC_KMS_KEY_ID)
+    _emit_audit(
+        "token.vend.allowed",
+        pat_jti=jti,
+        sub=oidc_claims["sub"],
+        policy_sha=policy_sha,
+        expires_at=oidc_claims["exp"],
+    )
+    return {"token": oidc_token, "expires_at": oidc_claims["exp"]}
+
+
+class _DeniedError(Exception):
+    """Raised on revocation / ABAC denial → 403."""
+
+    def __init__(self, reason: str):
+        self.reason = reason
+        super().__init__(f"token vend denied: {reason}")
+
+
+# ---------------------------------------------------------------------------
+# Lambda handler + HTTP mapping
+# ---------------------------------------------------------------------------
+
+
+def _to_http_response(result_or_error):
+    if isinstance(result_or_error, Exception):
+        if isinstance(result_or_error, _DeniedError):
+            return {
+                "statusCode": 403,
+                "body": json.dumps({"error": "token_vend_denied", "reason": result_or_error.reason}),
+            }
+        if isinstance(result_or_error, ValueError):
+            return {
+                "statusCode": 400,
+                "body": json.dumps({"error": str(result_or_error)}),
+            }
+        return {
+            "statusCode": 500,
+            "body": json.dumps({"error": str(result_or_error)}),
+        }
+    return {"statusCode": 200, "body": json.dumps(result_or_error)}
+
+
+def lambda_handler(event, context):
+    """AWS Lambda handler entry point (thin wrapper, REQ-329 dual-use)."""
+    try:
+        body = event.get("body", "{}")
+        payload = json.loads(body) if isinstance(body, str) else body
+        token = payload.get("token") or payload.get("pat") or payload.get("session_token")
+        if not token:
+            raise ValueError("missing field: token (or pat / session_token)")
+        result = vend_token(
+            token=token,
+            requested_claims=payload.get("requested_claims"),
+            target_resource=payload.get("target_resource"),
+            environment=payload.get("environment"),
+            policy_version=payload.get("policy_version", ""),
+        )
+        return _to_http_response(result)
+    except Exception as e:
+        return _to_http_response(e)
+
+
+# ---------------------------------------------------------------------------
+# CLI (dual-use, REQ-329 pattern)
+# ---------------------------------------------------------------------------
+
+
+def cli_main(argv=None):
+    """CLI entry point for the token-vend Lambda (REQ-329 dual-use)."""
+    raw = argv if argv is not None else sys.argv[1:]
+    local_bypass = os.environ.get("NOVA_LAMBDA_LOCAL_BYPASS")
+    if not local_bypass:
+        os.environ["NOVA_LAMBDA_LOCAL_BYPASS"] = "1"
+    try:
+        if "--vend-stdin" in raw:
+            payload = json.loads(sys.stdin.read())
+        elif "--vend" in raw:
+            idx = raw.index("--vend")
+            path = raw[idx + 1] if idx + 1 < len(raw) else None
+            if not path:
+                print("Usage: --vend <payload.json>", file=sys.stderr)
+                return 2
+            with open(path) as fh:
+                payload = json.loads(fh.read())
+        else:
+            print(
+                "Usage: python3 -m core.lambda.nova_idp_token_vend "
+                "--vend <payload.json> | --vend-stdin < <payload.json>",
+                file=sys.stderr,
+            )
+            return 2
+        token = payload.get("token") or payload.get("pat") or payload.get("session_token")
+        if not token:
+            print("error: missing token in payload", file=sys.stderr)
+            return 1
+        result = vend_token(
+            token=token,
+            requested_claims=payload.get("requested_claims"),
+            target_resource=payload.get("target_resource"),
+            environment=payload.get("environment"),
+            policy_version=payload.get("policy_version", ""),
+        )
+        sys.stdout.write(json.dumps(result, indent=2) + "\n")
+        return 0
+    except _DeniedError as e:
+        sys.stderr.write(f"error: token vend denied ({e.reason})\n")
+        return 3  # 403-class
+    except ValueError as e:
+        sys.stderr.write(f"error: {e}\n")
+        return 1
+    except Exception as e:  # pragma: no cover - defensive top-level guard
+        sys.stderr.write(f"internal error: {e}\n")
+        return 2
+    finally:
+        if not local_bypass:
+            os.environ.pop("NOVA_LAMBDA_LOCAL_BYPASS", None)
+
+
+if __name__ == "__main__":  # pragma: no cover - CLI entry
+    sys.exit(cli_main())
\ No newline at end of file
diff --git a/tests/test_abac_fail_closed.py b/tests/test_abac_fail_closed.py
new file mode 100644
index 0000000..21f00d6
--- /dev/null
+++ b/tests/test_abac_fail_closed.py
@@ -0,0 +1,231 @@
+"""ABAC fail-closed test for the token-vend Lambda (C-6.1/C-7.1, INV-17).
+
+🔴 THIS IS THE MOST IMPORTANT TEST OF THE MILESTONE. It verifies that
+INV-17 (ABAC fail-closed) is a **runtime guarantee**, not just
+documentation. The grill's #1 finding was that a naive implementation
+could fail open (vend a token when the ABAC engine is broken). This
+test pins the opposite: **every** ABAC failure mode → 403 +
+``token.vend.denied`` (reason ``abac_eval_failed``). Never fail open.
+
+Failure modes covered:
+  1. ``KyvernoJsonEngine.is_configured()`` returns ``False`` (kj absent).
+  2. ``evaluate_token_vend_policy()`` raises an exception (kj error,
+     policy parse error, subprocess crash).
+  3. ABAC denies (allowed=False) → 403 reason ``abac_denied``.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sys
+import time
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+# moto requires a region; the Lambda's lazy boto3.resource("dynamodb")
+# picks up AWS_DEFAULT_REGION.
+os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
+os.environ.setdefault("AWS_ACCESS_KEY_ID", "test")
+os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test")
+os.environ.setdefault("NOVA_LAMBDA_LOCAL_BYPASS", "1")
+
+# Load the token-vend Lambda via importlib (`lambda` is a reserved word).
+_SOURCE_PATH = (
+    Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_token_vend.py"
+)
+_spec = importlib.util.spec_from_file_location("nova_idp_token_vend", _SOURCE_PATH)
+tv = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(tv)
+
+# Load nova_idp_auth_cfn table helpers + moto for DDB.
+import boto3
+from moto import mock_aws
+
+
+def _create_pats_table(ddb):
+    ddb.create_table(
+        TableName="nova-pats",
+        KeySchema=[{"AttributeName": "jti", "KeyType": "HASH"}],
+        AttributeDefinitions=[
+            {"AttributeName": "jti", "AttributeType": "S"},
+            {"AttributeName": "sub", "AttributeType": "S"},
+            {"AttributeName": "pat_hash", "AttributeType": "S"},
+        ],
+        GlobalSecondaryIndexes=[
+            {"IndexName": "sub-index", "KeySchema": [{"AttributeName": "sub", "KeyType": "HASH"}], "Projection": {"ProjectionType": "ALL"}},
+            {"IndexName": "pat_hash-index", "KeySchema": [{"AttributeName": "pat_hash", "KeyType": "HASH"}], "Projection": {"ProjectionType": "ALL"}},
+        ],
+        BillingMode="PAY_PER_REQUEST",
+    )
+
+
+@pytest.fixture(autouse=True)
+def _reset_lambda_singletons():
+    """Reset the Lambda's module-level DynamoDB singleton before each test."""
+    tv._dynamodb = None
+    yield
+    tv._dynamodb = None
+
+
+def _put_active_pat(ddb, jti="pat-active", sub="user-1", owner="t1", role="developer"):
+    ddb.put_item(
+        TableName="nova-pats",
+        Item={
+            "jti": {"S": jti},
+            "sub": {"S": sub},
+            "pat_hash": {"S": "hash-" + jti},
+            "status": {"S": "active"},
+            "issued_at": {"S": "2026-01-01T00:00:00Z"},
+            "expires_at": {"N": str(int(time.time()) + 3600)},
+            "claims": {"S": json.dumps({"sub": sub, "roles": [role], "owner": owner})},
+        },
+    )
+
+
+def _make_pat_jwt(jti="pat-active", sub="user-1", role="developer", owner="t1"):
+    """Build an unsigned-ish JWT (signature irrelevant — decoded without verify)."""
+    import base64
+    header = base64.urlsafe_b64encode(json.dumps({"alg": "none", "typ": "JWT"}).encode()).rstrip(b"=").decode()
+    payload = base64.urlsafe_b64encode(json.dumps({
+        "jti": jti, "sub": sub, "exp": int(time.time()) + 3600,
+        "iat": int(time.time()), "roles": [role], "owner": owner,
+        "typ": "developer_pat",
+    }).encode()).rstrip(b"=").decode()
+    return f"{header}.{payload}.sig"
+
+
+def _vend_event(pat_jwt, **extra):
+    body = {"token": pat_jwt, "environment": "dev", "target_resource": {"type": "contract", "id": "c1", "owner": "t1", "environment": "dev"}, "requested_claims": ["sub"]}
+    body.update(extra)
+    return {"body": json.dumps(body)}
+
+
+# ---------------------------------------------------------------------------
+# 🔴 THE CRITICAL TESTS — fail closed on every ABAC failure mode.
+# ---------------------------------------------------------------------------
+
+
+@mock_aws
+def test_fail_closed_when_kj_not_configured():
+    """C-6.1: is_configured() == False → 403 + abac_eval_failed. NEVER fail open."""
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb)
+    pat = _make_pat_jwt()
+    # Mock the engine so is_configured() returns False (kj absent).
+    fake_engine = mock.MagicMock()
+    fake_engine.is_configured.return_value = False
+    with mock.patch("core.policy_engine.get_engine", return_value=fake_engine):
+        resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    body = json.loads(resp["body"])
+    assert body["reason"] == "abac_eval_failed"
+    assert body["error"] == "token_vend_denied"
+
+
+@mock_aws
+def test_fail_closed_when_evaluate_raises():
+    """C-6.1: evaluate() raises → 403 + abac_eval_failed. NEVER fail open."""
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb)
+    pat = _make_pat_jwt()
+    fake_engine = mock.MagicMock()
+    fake_engine.is_configured.return_value = True
+    # evaluate_token_vend_policy is called inside _evaluate_abac_fail_closed;
+    # patch the core.abac_evaluator module to raise.
+    with mock.patch("core.policy_engine.get_engine", return_value=fake_engine), \
+         mock.patch("core.abac_evaluator.evaluate_token_vend_policy",
+                    side_effect=RuntimeError("kj crashed")):
+        resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    body = json.loads(resp["body"])
+    assert body["reason"] == "abac_eval_failed"
+
+
+@mock_aws
+def test_fail_closed_when_policy_parse_error():
+    """C-6.1: policy parse error (evaluate raises ValueError) → 403."""
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb)
+    pat = _make_pat_jwt()
+    fake_engine = mock.MagicMock()
+    fake_engine.is_configured.return_value = True
+    with mock.patch("core.policy_engine.get_engine", return_value=fake_engine), \
+         mock.patch("core.abac_evaluator.evaluate_token_vend_policy",
+                    side_effect=ValueError("policy parse error")):
+        resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "abac_eval_failed"
+
+
+@mock_aws
+def test_fail_closed_when_abac_denies():
+    """ABAC denies (allowed=False) → 403 + abac_denied (distinct from eval_failed)."""
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb)
+    pat = _make_pat_jwt()
+    fake_engine = mock.MagicMock()
+    fake_engine.is_configured.return_value = True
+    with mock.patch("core.policy_engine.get_engine", return_value=fake_engine), \
+         mock.patch("core.abac_evaluator.evaluate_token_vend_policy",
+                    return_value=(False, [], "sha")):
+        resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "abac_denied"
+
+
+@mock_aws
+def test_fail_closed_revoked_pat():
+    """D-229: revoked PAT → 403 + pat_revoked (before ABAC even runs)."""
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb, jti="pat-rev")
+    ddb.update_item(
+        TableName="nova-pats",
+        Key={"jti": {"S": "pat-rev"}},
+        UpdateExpression="SET #s = :v",
+        ExpressionAttributeNames={"#s": "status"},
+        ExpressionAttributeValues={":v": {"S": "revoked"}},
+    )
+    pat = _make_pat_jwt(jti="pat-rev")
+    resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "pat_revoked"
+
+
+@mock_aws
+def test_fail_closed_unknown_pat():
+    """D-229: PAT not in table → 403 + pat_unknown."""
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    pat = _make_pat_jwt(jti="pat-missing")
+    resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "pat_unknown"
+
+
+@mock_aws
+def test_audit_event_emitted_on_denial(capsys):
+    """token.vend.denied audit event is emitted on every denial (INV-17)."""
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb)
+    pat = _make_pat_jwt()
+    fake_engine = mock.MagicMock()
+    fake_engine.is_configured.return_value = False
+    with mock.patch("core.policy_engine.get_engine", return_value=fake_engine):
+        tv.lambda_handler(_vend_event(pat), None)
+    err = capsys.readouterr().err
+    audit_lines = [l for l in err.strip().split("\n") if l.strip()]
+    denied = [json.loads(l) for l in audit_lines if json.loads(l).get("event") == "token.vend.denied"]
+    assert denied, "expected a token.vend.denied audit event"
+    assert denied[0]["reason"] == "abac_eval_failed"
\ No newline at end of file
diff --git a/tests/test_token_vend.py b/tests/test_token_vend.py
new file mode 100644
index 0000000..b559be9
--- /dev/null
+++ b/tests/test_token_vend.py
@@ -0,0 +1,195 @@
+"""E2E token-vend Lambda test (REQ-336, C-6.1) with moto + mock KMS.
+
+Valid PAT (active) + ABAC allow → KMS-signed OIDC token returned.
+Revoked PAT → 403. Unknown PAT → 403. ABAC deny → 403. The returned
+JWT verifies with pyjwt + the mock public key.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sys
+import time
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
+os.environ.setdefault("AWS_ACCESS_KEY_ID", "test")
+os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test")
+os.environ.setdefault("NOVA_LAMBDA_LOCAL_BYPASS", "1")
+
+_SOURCE_PATH = (
+    Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_token_vend.py"
+)
+_spec = importlib.util.spec_from_file_location("nova_idp_token_vend_e2e", _SOURCE_PATH)
+tv = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(tv)
+
+import boto3
+from moto import mock_aws
+import jwt as pyjwt
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives import hashes, serialization
+
+import core.kms_signing as kms_signing
+
+
+def _create_pats_table(ddb):
+    ddb.create_table(
+        TableName="nova-pats",
+        KeySchema=[{"AttributeName": "jti", "KeyType": "HASH"}],
+        AttributeDefinitions=[
+            {"AttributeName": "jti", "AttributeType": "S"},
+            {"AttributeName": "sub", "AttributeType": "S"},
+            {"AttributeName": "pat_hash", "AttributeType": "S"},
+        ],
+        GlobalSecondaryIndexes=[
+            {"IndexName": "sub-index", "KeySchema": [{"AttributeName": "sub", "KeyType": "HASH"}], "Projection": {"ProjectionType": "ALL"}},
+            {"IndexName": "pat_hash-index", "KeySchema": [{"AttributeName": "pat_hash", "KeyType": "HASH"}], "Projection": {"ProjectionType": "ALL"}},
+        ],
+        BillingMode="PAY_PER_REQUEST",
+    )
+
+
+def _put_active_pat(ddb, jti="pat-active", sub="user-1", owner="t1", role="developer"):
+    ddb.put_item(
+        TableName="nova-pats",
+        Item={
+            "jti": {"S": jti},
+            "sub": {"S": sub},
+            "pat_hash": {"S": "hash-" + jti},
+            "status": {"S": "active"},
+            "issued_at": {"S": "2026-01-01T00:00:00Z"},
+            "expires_at": {"N": str(int(time.time()) + 3600)},
+            "claims": {"S": json.dumps({"sub": sub, "roles": [role], "owner": owner})},
+        },
+    )
+
+
+def _make_pat_jwt(jti="pat-active", sub="user-1", role="developer", owner="t1"):
+    import base64
+    header = base64.urlsafe_b64encode(json.dumps({"alg": "none", "typ": "JWT"}).encode()).rstrip(b"=").decode()
+    payload = base64.urlsafe_b64encode(json.dumps({
+        "jti": jti, "sub": sub, "exp": int(time.time()) + 3600,
+        "iat": int(time.time()), "roles": [role], "owner": owner,
+        "typ": "developer_pat",
+    }).encode()).rstrip(b"=").decode()
+    return f"{header}.{payload}.sig"
+
+
+def _vend_event(pat_jwt, env="dev", owner="t1"):
+    return {"body": json.dumps({
+        "token": pat_jwt, "environment": env,
+        "target_resource": {"type": "contract", "id": "c1", "owner": owner, "environment": env},
+        "requested_claims": ["sub", "roles"],
+    })}
+
+
+class _MockKmsSign:
+    def __init__(self, priv, pub_der):
+        self._priv = priv
+        self._pub_der = pub_der
+
+    def sign(self, KeyId, Message, MessageType, SigningAlgorithm):
+        return {"Signature": self._priv.sign(Message, ec.ECDSA(hashes.SHA256()))}
+
+    def get_public_key(self, KeyId):
+        return {"PublicKey": self._pub_der}
+
+
+@pytest.fixture
+def test_keypair():
+    priv = ec.generate_private_key(ec.SECP256R1())
+    pub = priv.public_key()
+    pub_der = pub.public_bytes(
+        encoding=serialization.Encoding.DER,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    return priv, pub, pub_der
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    tv._dynamodb = None
+    yield
+    tv._dynamodb = None
+    kms_signing.set_kms_client_for_testing(None)
+
+
+@mock_aws
+def test_valid_pat_abac_allow_vends_token(test_keypair):
+    priv, pub, pub_der = test_keypair
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb)
+    kms_signing.set_kms_client_for_testing(_MockKmsSign(priv, pub_der))
+    pat = _make_pat_jwt()
+    # ABAC allow: developer + dev
+    resp = tv.lambda_handler(_vend_event(pat, env="dev", owner="t1"), None)
+    assert resp["statusCode"] == 200, resp["body"]
+    body = json.loads(resp["body"])
+    assert "token" in body
+    assert "expires_at" in body
+    # Verify the JWT with pyjwt + the test public key.
+    pem = pub.public_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    ).decode("ascii")
+    decoded = pyjwt.decode(body["token"], pem, algorithms=["ES256"], options={"verify_aud": False})
+    assert decoded["sub"] == "user-1"
+    assert decoded["typ"] == "nova_oidc_token"
+    assert decoded["roles"] == ["developer"]
+    assert "jti" in decoded and "iat" in decoded and "exp" in decoded and "iss" in decoded
+
+
+@mock_aws
+def test_revoked_pat_denied():
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb, jti="pat-r")
+    ddb.update_item(
+        TableName="nova-pats", Key={"jti": {"S": "pat-r"}},
+        UpdateExpression="SET #s = :v",
+        ExpressionAttributeNames={"#s": "status"},
+        ExpressionAttributeValues={":v": {"S": "revoked"}},
+    )
+    pat = _make_pat_jwt(jti="pat-r")
+    resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "pat_revoked"
+
+
+@mock_aws
+def test_unknown_pat_denied():
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    pat = _make_pat_jwt(jti="pat-missing")
+    resp = tv.lambda_handler(_vend_event(pat), None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "pat_unknown"
+
+
+@mock_aws
+def test_abac_deny_denied(test_keypair):
+    priv, _pub, pub_der = test_keypair
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    _put_active_pat(ddb)
+    kms_signing.set_kms_client_for_testing(_MockKmsSign(priv, pub_der))
+    pat = _make_pat_jwt()
+    # ABAC deny: developer + prod (developer not allowed in prod)
+    resp = tv.lambda_handler(_vend_event(pat, env="prod", owner="t1"), None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "abac_denied"
+
+
+@mock_aws
+def test_missing_token_field():
+    resp = tv.lambda_handler({"body": json.dumps({"environment": "dev"})}, None)
+    assert resp["statusCode"] == 400
\ No newline at end of file