diff --git a/examples/run_task_integration.py b/examples/run_task_integration.py
new file mode 100644
index 00000000..71b43b8c
--- /dev/null
+++ b/examples/run_task_integration.py
@@ -0,0 +1,130 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+"""Example Run Task callback integration.
+
+This example sends a callback result back to Terraform after a Run Task
+webhook is received.
+
+Required environment variables:
+
+- TFE_ADDRESS
+    Terraform address (for example: https://app.terraform.io)
+
+- TFE_TOKEN
+    Your Terraform API token used to initialize the SDK client.
+
+- TFE_CALLBACK_URL
+    The task_result_callback_url received in the Run Task webhook payload.
+
+- TFE_CALLBACK_TOKEN
+    The access_token received in the same webhook payload.
+    This token is used for the callback request and is different from
+    your regular Terraform API token.
+
+Local testing flow:
+
+1. Start the webhook server:
+
+       uvicorn examples.run_task_webhook_server:app --reload --port 8000
+
+2. Expose the server publicly:
+
+       ngrok http 8000
+
+3. Create a Run Task in Terraform Cloud / Enterprise using the ngrok URL.
+
+4. Attach the Run Task to a workspace and trigger a run.
+
+5. The webhook payload will include values similar to:
+
+       {
+         "task_result_callback_url": "https://app.terraform.io/...",
+         "access_token": "v1.xxxxx..."
+       }
+
+6. Export those values locally and run this example script,
+   or call client.run_task_integrations.callback(...) directly
+   inside your webhook handler.
+
+Example:
+
+    export TFE_ADDRESS=https://app.terraform.io
+    export TFE_TOKEN=<your-api-token>
+    export TFE_CALLBACK_URL=<from-webhook-payload>
+    export TFE_CALLBACK_TOKEN=<from-webhook-payload>
+
+    python examples/run_task_integration.py
+"""
+
+from __future__ import annotations
+
+import os
+
+from pytfe import TFEClient, TFEConfig
+from pytfe.models.run_task_integration import (
+    TaskResultCallbackRequestOptions,
+    TaskResultOutcome,
+    TaskResultStatus,
+    TaskResultTag,
+)
+
+
+def main() -> None:
+    callback_url = os.getenv("TFE_CALLBACK_URL")
+    access_token = os.getenv("TFE_CALLBACK_TOKEN")
+
+    if not callback_url or not access_token:
+        print("Missing TFE_CALLBACK_URL or TFE_CALLBACK_TOKEN")
+        return
+
+    # TFE_ADDRESS and TFE_TOKEN are loaded from the environment.
+    # The callback request itself uses the short-lived webhook token.
+    client = TFEClient(TFEConfig.from_env())
+
+    outcome = TaskResultOutcome(
+        description="Example outcome",
+        body="All checks passed successfully",
+        tags={"severity": [TaskResultTag(label="low", level="info")]},
+    )
+
+    # Example status values:
+    #
+    # - passed: marks the run task as successful
+    # - failed: fails the run task
+    # - running: reports progress before sending a final result
+    #
+    # Example: send an in-progress update
+    #
+    # options = TaskResultCallbackRequestOptions(
+    #     status=TaskResultStatus.running,
+    #     message="Security scan in progress",
+    # )
+    #
+    # Example: report a failure
+    #
+    # options = TaskResultCallbackRequestOptions(
+    #     status=TaskResultStatus.failed,
+    #     message="Found critical vulnerabilities",
+    # )
+
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        message="Run task completed successfully",
+        url="https://example.com/results",
+        outcomes=[outcome],
+    )
+
+    print(f"Sending callback to: {callback_url}")
+
+    client.run_task_integrations.callback(
+        callback_url=callback_url,
+        access_token=access_token,
+        options=options,
+    )
+
+    print("Run task callback sent successfully")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/run_task_webhook_server.py b/examples/run_task_webhook_server.py
new file mode 100644
index 00000000..5bb09797
--- /dev/null
+++ b/examples/run_task_webhook_server.py
@@ -0,0 +1,103 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+"""Minimal FastAPI webhook server for Terraform Run Tasks.
+
+This example receives a Run Task webhook, extracts the callback URL
+and access token from the payload, and sends a callback result back
+to Terraform using the SDK.
+
+Setup:
+
+1. Install dependencies:
+
+       pip install fastapi uvicorn
+
+2. Configure environment variables:
+
+       export TFE_ADDRESS=https://app.terraform.io
+       export TFE_TOKEN=<your-api-token>
+
+3. Start the server from the repository root:
+
+       uvicorn examples.run_task_webhook_server:app --reload --port 8000
+
+4. Expose the server publicly with ngrok:
+
+       ngrok http 8000
+
+5. In Terraform Cloud / Enterprise:
+
+   - Create a Run Task using the ngrok URL
+   - Attach the Run Task to a workspace
+   - Trigger a Terraform run
+
+The webhook payload will include values like:
+
+       {
+         "task_result_callback_url": "...",
+         "access_token": "..."
+       }
+
+This example prints the payload locally and sends a successful
+callback response back to Terraform.
+"""
+
+from __future__ import annotations
+
+import json
+
+from fastapi import FastAPI, Request
+
+from pytfe import TFEClient, TFEConfig
+from pytfe.models.run_task_integration import (
+    TaskResultCallbackRequestOptions,
+    TaskResultStatus,
+)
+
+app = FastAPI()
+client = TFEClient(TFEConfig.from_env())
+
+
+@app.post("/")
+async def receive_webhook(request: Request) -> dict[str, bool]:
+    try:
+        payload = await request.json()
+    except Exception:
+        # Terraform verification requests may not include a JSON payload.
+        return {"ok": True}
+
+    print("\n=== FULL PAYLOAD ===")
+    print(json.dumps(payload, indent=2))
+
+    callback_url = payload.get("task_result_callback_url")
+    access_token = payload.get("access_token")
+
+    print("\n=== EXTRACTED VALUES ===")
+    print("callback_url:", callback_url)
+    print("access_token:", access_token)
+
+    if not callback_url or not access_token:
+        # Verification requests do not include callback information.
+        return {"ok": True}
+
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        message="Webhook received and processed",
+        url="https://github.com/hashicorp/python-tfe",
+    )
+
+    print(f"Sending callback to: {callback_url}")
+
+    try:
+        client.run_task_integrations.callback(
+            callback_url=callback_url,
+            access_token=access_token,
+            options=options,
+        )
+        print("Run task callback sent successfully")
+    except Exception as exc:
+        print(f"Callback failed: {exc!r}")
+        return {"ok": False}
+
+    return {"ok": True}
diff --git a/src/pytfe/client.py b/src/pytfe/client.py
index 4642d9a8..59a8c47f 100644
--- a/src/pytfe/client.py
+++ b/src/pytfe/client.py
@@ -33,6 +33,7 @@
 from .resources.run import Runs
 from .resources.run_event import RunEvents
 from .resources.run_task import RunTasks
+from .resources.run_task_integration import RunTaskIntegrations
 from .resources.run_trigger import RunTriggers
 from .resources.ssh_keys import SSHKeys
 from .resources.stack import Stacks
@@ -100,6 +101,7 @@ def __init__(self, config: TFEConfig | None = None):
         self.state_versions = StateVersions(self._transport)
         self.state_version_outputs = StateVersionOutputs(self._transport)
         self.run_tasks = RunTasks(self._transport)
+        self.run_task_integrations = RunTaskIntegrations(self._transport)
         self.run_triggers = RunTriggers(self._transport)
         self.runs = Runs(self._transport)
         self.query_runs = QueryRuns(self._transport)
diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py
index 113dee9a..73b0e848 100644
--- a/src/pytfe/errors.py
+++ b/src/pytfe/errors.py
@@ -58,6 +58,24 @@ class RequiredFieldMissing(TFEError): ...
 class ErrStateVersionUploadNotSupported(TFEError): ...
 
 
+class InvalidCallbackURLError(TFEError):
+    def __init__(self, message: str = "Invalid callback URL") -> None:
+        super().__init__(message)
+
+
+class InvalidAccessTokenError(TFEError):
+    def __init__(self, message: str = "Invalid access token") -> None:
+        super().__init__(message)
+
+
+class InvalidTaskResultsCallbackStatusError(TFEError):
+    def __init__(
+        self,
+        message: str = "Invalid task result callback status; must be one of: passed, failed, running",
+    ) -> None:
+        super().__init__(message)
+
+
 # Generic error constants
 ERR_UNAUTHORIZED = "unauthorized"
 ERR_RESOURCE_NOT_FOUND = "resource not found"
diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py
index d2e8648c..3aafb44f 100644
--- a/src/pytfe/models/__init__.py
+++ b/src/pytfe/models/__init__.py
@@ -296,6 +296,14 @@
     Stage,
     TaskEnforcementLevel,
 )
+from .run_task_integration import (
+    TaskResultCallbackRequestOptions,
+    TaskResultOutcome,
+    TaskResultTag,
+)
+from .run_task_integration import (
+    TaskResultStatus as TaskResultCallbackStatus,
+)
 from .run_trigger import (
     RunTrigger,
     RunTriggerCreateOptions,
@@ -654,6 +662,11 @@
     "RunTaskCreateOptions",
     "RunTaskUpdateOptions",
     "RunTaskReadOptions",
+    # Run task integration (callback)
+    "TaskResultCallbackRequestOptions",
+    "TaskResultCallbackStatus",
+    "TaskResultOutcome",
+    "TaskResultTag",
     # Run triggers
     "RunTrigger",
     "RunTriggerCreateOptions",
diff --git a/src/pytfe/models/run_task_integration.py b/src/pytfe/models/run_task_integration.py
new file mode 100644
index 00000000..ee3b4c00
--- /dev/null
+++ b/src/pytfe/models/run_task_integration.py
@@ -0,0 +1,108 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from ..errors import InvalidTaskResultsCallbackStatusError
+
+
+class TaskResultStatus(str, Enum):
+    """Statuses accepted by the Run Task callback endpoint.
+
+    Mirrors the Go SDK's accepted callback statuses (passed, failed, running).
+    """
+
+    passed = "passed"
+    failed = "failed"
+    running = "running"
+
+
+class TaskResultTag(BaseModel):
+    """Tag attached to a Run Task outcome to enrich the result display in the UI."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    label: str = Field(..., alias="label")
+    level: str | None = Field(None, alias="level")
+
+    def _to_payload(self) -> dict[str, Any]:
+        payload: dict[str, Any] = {"label": self.label}
+        if self.level is not None:
+            payload["level"] = self.level
+        return payload
+
+
+class TaskResultOutcome(BaseModel):
+    """Detailed Run Task outcome."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    outcome_id: str | None = Field(None, alias="outcome-id")
+    description: str | None = Field(None, alias="description")
+    body: str | None = Field(None, alias="body")
+    url: str | None = Field(None, alias="url")
+    tags: dict[str, list[TaskResultTag]] | None = Field(None, alias="tags")
+
+    def _to_payload(self) -> dict[str, Any]:
+        attributes: dict[str, Any] = {}
+        if self.outcome_id is not None:
+            attributes["outcome-id"] = self.outcome_id
+        if self.description is not None:
+            attributes["description"] = self.description
+        if self.body is not None:
+            attributes["body"] = self.body
+        if self.url is not None:
+            attributes["url"] = self.url
+        if self.tags is not None:
+            attributes["tags"] = {
+                key: [tag._to_payload() for tag in tags]
+                for key, tags in self.tags.items()
+            }
+        return {"type": "task-result-outcomes", "attributes": attributes}
+
+
+class TaskResultCallbackRequestOptions(BaseModel):
+    """Payload options for sending a Run Task callback result."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    status: TaskResultStatus = Field(..., alias="status")
+    message: str | None = Field(None, alias="message")
+    url: str | None = Field(None, alias="url")
+    outcomes: list[TaskResultOutcome] | None = Field(None, alias="outcomes")
+
+    def _validate(self) -> None:
+        """Validate callback status."""
+        if not isinstance(self.status, TaskResultStatus):
+            raise InvalidTaskResultsCallbackStatusError()
+
+    def to_payload(self) -> dict[str, Any]:
+        """Return the JSON:API payload for the callback PATCH request."""
+        self._validate()
+
+        attributes: dict[str, Any] = {"status": self.status.value}
+        if self.message is not None:
+            attributes["message"] = self.message
+        if self.url is not None:
+            attributes["url"] = self.url
+
+        payload: dict[str, Any] = {
+            "data": {
+                "type": "task-results",
+                "attributes": attributes,
+            }
+        }
+
+        if self.outcomes:
+            payload["data"]["relationships"] = {
+                "outcomes": {
+                    "data": [outcome._to_payload() for outcome in self.outcomes]
+                }
+            }
+
+        return payload
diff --git a/src/pytfe/resources/run_task_integration.py b/src/pytfe/resources/run_task_integration.py
new file mode 100644
index 00000000..165dbbf2
--- /dev/null
+++ b/src/pytfe/resources/run_task_integration.py
@@ -0,0 +1,42 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+from __future__ import annotations
+
+from ..errors import InvalidAccessTokenError, InvalidCallbackURLError
+from ..models.run_task_integration import TaskResultCallbackRequestOptions
+from ._base import _Service
+
+
+class RunTaskIntegrations(_Service):
+    """Run Tasks Integration Callback API.
+
+    See:
+    https://developer.hashicorp.com/terraform/enterprise/api-docs/run-tasks/run-tasks-integration
+    """
+
+    def callback(
+        self,
+        callback_url: str,
+        access_token: str,
+        options: TaskResultCallbackRequestOptions,
+    ) -> None:
+        """Send a Run Task result back to the Terraform callback URL.
+
+        The PATCH request must use the access token from the originating
+        Run Task webhook (not the SDK client's API token).
+        """
+        if not callback_url or not callback_url.strip():
+            raise InvalidCallbackURLError()
+        if not access_token or not access_token.strip():
+            raise InvalidAccessTokenError()
+
+        self.t.request(
+            "PATCH",
+            callback_url,
+            json_body=options.to_payload(),
+            headers={
+                "Authorization": f"Bearer {access_token}",
+                "Content-Type": "application/vnd.api+json",
+            },
+        )
diff --git a/tests/units/test_run_task_integration.py b/tests/units/test_run_task_integration.py
new file mode 100644
index 00000000..f5145efc
--- /dev/null
+++ b/tests/units/test_run_task_integration.py
@@ -0,0 +1,516 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+import pytest
+
+from pytfe._http import HTTPTransport
+from pytfe.errors import (
+    InvalidAccessTokenError,
+    InvalidCallbackURLError,
+    InvalidTaskResultsCallbackStatusError,
+    TFEError,
+)
+from pytfe.models.run_task_integration import (
+    TaskResultCallbackRequestOptions,
+    TaskResultOutcome,
+    TaskResultStatus,
+    TaskResultTag,
+)
+from pytfe.resources._base import _Service
+from pytfe.resources.run_task_integration import RunTaskIntegrations
+
+CALLBACK_URL = "https://app.terraform.io/api/v2/task-results/taskrs-abc/callback"
+ACCESS_TOKEN = "v1.callback-token"
+
+
+@pytest.fixture
+def transport() -> Mock:
+    return Mock(spec=HTTPTransport)
+
+
+@pytest.fixture
+def service(transport: Mock) -> RunTaskIntegrations:
+    return RunTaskIntegrations(transport)
+
+
+def _basic_options() -> TaskResultCallbackRequestOptions:
+    return TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        message="All good",
+        url="https://example.com/details",
+    )
+
+
+# ─── Architectural sanity ─────────────────────────────────────────────────────
+
+
+def test_service_extends_base_service():
+    assert issubclass(RunTaskIntegrations, _Service)
+
+
+def test_service_uses_transport(transport, service):
+    assert service.t is transport
+
+
+def test_typed_errors_subclass_tfe_error():
+    assert issubclass(InvalidCallbackURLError, TFEError)
+    assert issubclass(InvalidAccessTokenError, TFEError)
+    assert issubclass(InvalidTaskResultsCallbackStatusError, TFEError)
+
+
+# ─── Validation: callback URL ─────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("bad_url", ["", "   ", "\t\n", None])
+def test_callback_invalid_url_raises_typed_error(service, bad_url):
+    with pytest.raises(InvalidCallbackURLError):
+        service.callback(bad_url, ACCESS_TOKEN, _basic_options())  # type: ignore[arg-type]
+
+
+# ─── Validation: access token ─────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("bad_token", ["", "   ", "\t\n", None])
+def test_callback_invalid_token_raises_typed_error(service, bad_token):
+    with pytest.raises(InvalidAccessTokenError):
+        service.callback(CALLBACK_URL, bad_token, _basic_options())  # type: ignore[arg-type]
+
+
+# ─── Validation: status ───────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "good_status",
+    [TaskResultStatus.passed, TaskResultStatus.failed, TaskResultStatus.running],
+)
+def test_callback_accepts_all_valid_statuses(service, transport, good_status):
+    options = TaskResultCallbackRequestOptions(status=good_status)
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    attrs = transport.request.call_args.kwargs["json_body"]["data"]["attributes"]
+    assert attrs["status"] == good_status.value
+
+
+@pytest.mark.parametrize(
+    "bad_status",
+    ["pending", "errored", "unreachable", "", "PASSED", "unknown", None, 123],
+)
+def test_callback_rejects_invalid_statuses(service, bad_status):
+    options = TaskResultCallbackRequestOptions(status=TaskResultStatus.passed)
+    options.status = bad_status  # type: ignore[assignment]
+    with pytest.raises(InvalidTaskResultsCallbackStatusError):
+        service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+
+
+# ─── Transport invocation ─────────────────────────────────────────────────────
+
+
+def test_callback_invokes_transport_with_exact_args(service, transport):
+    expected_payload = {
+        "data": {
+            "type": "task-results",
+            "attributes": {
+                "status": "passed",
+                "message": "All good",
+                "url": "https://example.com/details",
+            },
+        }
+    }
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options())
+
+    transport.request.assert_called_once_with(
+        "PATCH",
+        CALLBACK_URL,
+        json_body=expected_payload,
+        headers={
+            "Authorization": f"Bearer {ACCESS_TOKEN}",
+            "Content-Type": "application/vnd.api+json",
+        },
+    )
+
+
+def test_callback_passes_absolute_url_unchanged(service, transport):
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options())
+    args, _ = transport.request.call_args
+    assert args[0] == "PATCH"
+    assert args[1] == CALLBACK_URL
+    assert args[1].startswith("https://")
+
+
+def test_authorization_header_uses_callback_token(service, transport):
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options())
+    kwargs = transport.request.call_args.kwargs
+    assert kwargs["headers"] == {
+        "Authorization": f"Bearer {ACCESS_TOKEN}",
+        "Content-Type": "application/vnd.api+json",
+    }
+
+
+def test_callback_does_not_call_transport_on_validation_failure(service, transport):
+    with pytest.raises(InvalidCallbackURLError):
+        service.callback("", ACCESS_TOKEN, _basic_options())
+    transport.request.assert_not_called()
+
+
+# ─── Payload serialization: exact JSON:API shape ──────────────────────────────
+
+
+def test_payload_basic_exact_shape(service, transport):
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options())
+    assert transport.request.call_args.kwargs["json_body"] == {
+        "data": {
+            "type": "task-results",
+            "attributes": {
+                "status": "passed",
+                "message": "All good",
+                "url": "https://example.com/details",
+            },
+        }
+    }
+
+
+def test_payload_status_only_exact_shape(service, transport):
+    options = TaskResultCallbackRequestOptions(status=TaskResultStatus.running)
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    assert transport.request.call_args.kwargs["json_body"] == {
+        "data": {
+            "type": "task-results",
+            "attributes": {"status": "running"},
+        }
+    }
+
+
+def test_payload_with_outcomes_and_tags_exact_shape(service, transport):
+    outcome = TaskResultOutcome(
+        outcome_id="o-1",
+        description="desc",
+        body="body",
+        url="https://example.com/o1",
+        tags={
+            "severity": [
+                TaskResultTag(label="high", level="error"),
+                TaskResultTag(label="cve"),
+            ]
+        },
+    )
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.failed, outcomes=[outcome]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+
+    assert transport.request.call_args.kwargs["json_body"] == {
+        "data": {
+            "type": "task-results",
+            "attributes": {"status": "failed"},
+            "relationships": {
+                "outcomes": {
+                    "data": [
+                        {
+                            "type": "task-result-outcomes",
+                            "attributes": {
+                                "outcome-id": "o-1",
+                                "description": "desc",
+                                "body": "body",
+                                "url": "https://example.com/o1",
+                                "tags": {
+                                    "severity": [
+                                        {"label": "high", "level": "error"},
+                                        {"label": "cve"},
+                                    ]
+                                },
+                            },
+                        }
+                    ]
+                }
+            },
+        }
+    }
+
+
+# ─── Omission (`omitempty` parity) ────────────────────────────────────────────
+
+
+def test_message_omitted_when_none(service, transport):
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, url="https://x"
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    attrs = transport.request.call_args.kwargs["json_body"]["data"]["attributes"]
+    assert "message" not in attrs
+
+
+def test_url_omitted_when_none(service, transport):
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, message="m"
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    attrs = transport.request.call_args.kwargs["json_body"]["data"]["attributes"]
+    assert "url" not in attrs
+
+
+def test_relationships_omitted_when_outcomes_none(service, transport):
+    options = TaskResultCallbackRequestOptions(status=TaskResultStatus.passed)
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    body = transport.request.call_args.kwargs["json_body"]
+    assert "relationships" not in body["data"]
+
+
+def test_relationships_omitted_when_outcomes_empty_list(service, transport):
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, outcomes=[]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    body = transport.request.call_args.kwargs["json_body"]
+    assert "relationships" not in body["data"]
+
+
+def test_outcome_attributes_omit_none_fields(service, transport):
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, outcomes=[TaskResultOutcome()]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    entry = transport.request.call_args.kwargs["json_body"]["data"]["relationships"][
+        "outcomes"
+    ]["data"][0]
+    assert entry == {"type": "task-result-outcomes", "attributes": {}}
+
+
+def test_tag_level_omitted_when_none(service, transport):
+    outcome = TaskResultOutcome(tags={"category": [TaskResultTag(label="only")]})
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, outcomes=[outcome]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    tags = transport.request.call_args.kwargs["json_body"]["data"]["relationships"][
+        "outcomes"
+    ]["data"][0]["attributes"]["tags"]
+    assert tags == {"category": [{"label": "only"}]}
+
+
+def test_outcome_tags_omitted_when_none(service, transport):
+    outcome = TaskResultOutcome(description="no tags here")
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, outcomes=[outcome]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    attrs = transport.request.call_args.kwargs["json_body"]["data"]["relationships"][
+        "outcomes"
+    ]["data"][0]["attributes"]
+    assert attrs == {"description": "no tags here"}
+
+
+# ─── Edge cases ───────────────────────────────────────────────────────────────
+
+
+def test_multiple_outcomes_preserve_order(service, transport):
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        outcomes=[
+            TaskResultOutcome(outcome_id="o-1", description="first"),
+            TaskResultOutcome(outcome_id="o-2", description="second"),
+            TaskResultOutcome(outcome_id="o-3", description="third"),
+        ],
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    outcomes = transport.request.call_args.kwargs["json_body"]["data"]["relationships"][
+        "outcomes"
+    ]["data"]
+    assert [o["attributes"]["outcome-id"] for o in outcomes] == ["o-1", "o-2", "o-3"]
+
+
+def test_multiple_tags_per_category(service, transport):
+    outcome = TaskResultOutcome(
+        tags={
+            "severity": [
+                TaskResultTag(label="critical", level="error"),
+                TaskResultTag(label="high", level="error"),
+                TaskResultTag(label="medium", level="warning"),
+            ],
+            "compliance": [TaskResultTag(label="pci-dss")],
+        }
+    )
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.failed, outcomes=[outcome]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    tags = transport.request.call_args.kwargs["json_body"]["data"]["relationships"][
+        "outcomes"
+    ]["data"][0]["attributes"]["tags"]
+    assert tags == {
+        "severity": [
+            {"label": "critical", "level": "error"},
+            {"label": "high", "level": "error"},
+            {"label": "medium", "level": "warning"},
+        ],
+        "compliance": [{"label": "pci-dss"}],
+    }
+
+
+def test_unicode_message_and_body(service, transport):
+    outcome = TaskResultOutcome(body="✓ all good — 通过")
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        message="résumé 🎉",
+        outcomes=[outcome],
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    body = transport.request.call_args.kwargs["json_body"]
+    assert body["data"]["attributes"]["message"] == "résumé 🎉"
+    assert (
+        body["data"]["relationships"]["outcomes"]["data"][0]["attributes"]["body"]
+        == "✓ all good — 通过"
+    )
+
+
+def test_markdown_body_preserved_verbatim(service, transport):
+    md = "## Results\n\n- [link](https://x)\n- **bold**\n\n```py\nprint('ok')\n```"
+    outcome = TaskResultOutcome(body=md)
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, outcomes=[outcome]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    serialized = transport.request.call_args.kwargs["json_body"]["data"][
+        "relationships"
+    ]["outcomes"]["data"][0]["attributes"]["body"]
+    assert serialized == md
+
+
+def test_status_serialized_as_plain_string(service, transport):
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options())
+    status = transport.request.call_args.kwargs["json_body"]["data"]["attributes"][
+        "status"
+    ]
+    assert isinstance(status, str)
+    assert status == "passed"
+
+
+# ─── Pydantic alias / model behavior ──────────────────────────────────────────
+
+
+def test_outcome_accepts_alias_input():
+    outcome = TaskResultOutcome.model_validate({"outcome-id": "o-1"})
+    assert outcome.outcome_id == "o-1"
+
+
+def test_options_accepts_string_status():
+    options = TaskResultCallbackRequestOptions.model_validate({"status": "passed"})
+    assert options.status == TaskResultStatus.passed
+
+
+# ─── SDK client wiring ────────────────────────────────────────────────────────
+
+
+def test_client_wires_run_task_integrations():
+    """The TFEClient must expose `run_task_integrations` as a RunTaskIntegrations
+    bound to the client transport. Catches accidental rename / unwiring."""
+    from pytfe import TFEClient, TFEConfig
+
+    client = TFEClient(
+        TFEConfig(address="https://app.terraform.io", token="dummy-token")
+    )
+    assert isinstance(client.run_task_integrations, RunTaskIntegrations)
+    assert client.run_task_integrations.t is client._transport
+
+
+# ─── Return value & idempotency ───────────────────────────────────────────────
+
+
+def test_callback_returns_none(service, transport):
+    transport.request.return_value = {"data": {"id": "ignored"}}
+    assert service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options()) is None
+
+
+def test_to_payload_is_idempotent():
+    """Calling to_payload twice must produce equal dicts and must not mutate
+    the options instance — important because callers may inspect/log payloads."""
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        message="hi",
+        outcomes=[
+            TaskResultOutcome(
+                outcome_id="o-1",
+                tags={"sev": [TaskResultTag(label="high", level="error")]},
+            )
+        ],
+    )
+    first = options.to_payload()
+    second = options.to_payload()
+    assert first == second
+    # Mutating the returned payload must not affect the next serialization.
+    first["data"]["attributes"]["status"] = "mutated"
+    assert options.to_payload()["data"]["attributes"]["status"] == "passed"
+
+
+def test_to_payload_is_json_serializable():
+    """The transport ultimately json.dumps the body; the payload must contain
+    only JSON-native types (no Enum, no Pydantic models)."""
+    import json
+
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.failed,
+        outcomes=[
+            TaskResultOutcome(
+                outcome_id="o-1",
+                tags={"sev": [TaskResultTag(label="high", level="error")]},
+            )
+        ],
+    )
+    encoded = json.dumps(options.to_payload())
+    assert json.loads(encoded) == options.to_payload()
+
+
+# ─── Transport-side errors ────────────────────────────────────────────────────
+
+
+def test_transport_exception_propagates(service, transport):
+    """If the transport raises (e.g. network/HTTP error), the SDK must not
+    swallow it — callers need the failure to retry/log."""
+    transport.request.side_effect = RuntimeError("boom")
+    with pytest.raises(RuntimeError, match="boom"):
+        service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options())
+
+
+def test_sequential_callbacks_are_independent(service, transport):
+    """Two callbacks on the same service must produce two distinct requests."""
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, _basic_options())
+    service.callback(
+        CALLBACK_URL,
+        "v1.other-token",
+        TaskResultCallbackRequestOptions(status=TaskResultStatus.failed),
+    )
+    assert transport.request.call_count == 2
+    assert transport.request.call_args_list[0].kwargs["headers"] == {
+        "Authorization": f"Bearer {ACCESS_TOKEN}",
+        "Content-Type": "application/vnd.api+json",
+    }
+    assert transport.request.call_args_list[1].kwargs["headers"] == {
+        "Authorization": "Bearer v1.other-token",
+        "Content-Type": "application/vnd.api+json",
+    }
+    assert (
+        transport.request.call_args_list[1].kwargs["json_body"]["data"]["attributes"][
+            "status"
+        ]
+        == "failed"
+    )
+
+
+# ─── Current-behavior pins for empty-collection edge cases ────────────────────
+
+
+def test_outcome_with_empty_tags_dict_emits_empty_object(service, transport):
+    """Document current behavior: tags={} serializes as an empty object rather
+    than being omitted. Go SDK ``omitempty`` would drop it; if parity is
+    desired later, update both the model and this test together."""
+    outcome = TaskResultOutcome(tags={})
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed, outcomes=[outcome]
+    )
+    service.callback(CALLBACK_URL, ACCESS_TOKEN, options)
+    attrs = transport.request.call_args.kwargs["json_body"]["data"]["relationships"][
+        "outcomes"
+    ]["data"][0]["attributes"]
+    assert attrs == {"tags": {}}