AWS AppSync GraphQL Subscriptions: Push Live Coding-Agent Progress Without Polling

# ✅ keep payloads small — UI can refetch heavy diffs via query
enum ProgressPhase {
  PLANNED
  TOOL_STARTED
  TOOL_FINISHED
  DIFF_READY
  FAILED
  COMPLETED
}

type AgentProgress @aws_cognito_user_pools @aws_iam {
  runId: ID!
  tenantId: ID!
  phase: ProgressPhase!
  toolName: String
  message: String
  percent: Int
  artifactKey: String
  ts: AWSDateTime!
}

type Mutation {
  publishAgentProgress(input: PublishProgressInput!): AgentProgress
    @aws_iam
}

type Subscription {
  onAgentProgress(runId: ID!): AgentProgress
    @aws_subscribe(mutations: ["publishAgentProgress"])
}

input PublishProgressInput {
  runId: ID!
  tenantId: ID!
  phase: ProgressPhase!
  toolName: String
  message: String
  percent: Int
  artifactKey: String
}

# ✅ Lambda / Fargate worker — IAM auth to AppSync
import json, os, datetime
from urllib import request
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

APPSYNC = os.environ["APPSYNC_URL"]  # https://xxxx.appsync-api.region.amazonaws.com/graphql

MUTATION = """
mutation Publish($input: PublishProgressInput!) {
  publishAgentProgress(input: $input) {
    runId phase toolName message percent ts
  }
}
"""

def publish_progress(run_id: str, tenant_id: str, phase: str, **extra):
    payload = {
        "query": MUTATION,
        "variables": {
            "input": {
                "runId": run_id,
                "tenantId": tenant_id,
                "phase": phase,
                "ts": datetime.datetime.utcnow().isoformat() + "Z",
                **{k: v for k, v in extra.items() if v is not None},
            }
        },
    }
    body = json.dumps(payload).encode()
    aws_req = AWSRequest(method="POST", url=APPSYNC, data=body, headers={"Content-Type": "application/json"})
    SigV4Auth(boto3.Session().get_credentials(), "appsync", os.environ["AWS_REGION"]).add_auth(aws_req)
    # ❌ publishing without SigV4 / with a shared API key in the sandbox image
    http_req = request.Request(APPSYNC, data=body, headers=dict(aws_req.headers), method="POST")
    with request.urlopen(http_req, timeout=10) as resp:
        out = json.loads(resp.read().decode())
    if out.get("errors"):
        raise RuntimeError(out["errors"])
    return out["data"]["publishAgentProgress"]

// ✅ React — subscribe filtered by runId; Cognito JWT via Amplify
import { generateClient } from "aws-amplify/api";

const client = generateClient();

const sub = client
  .graphql({
    query: /* GraphQL */ `
      subscription OnProgress($runId: ID!) {
        onAgentProgress(runId: $runId) {
          phase toolName message percent artifactKey ts
        }
      }
    `,
    variables: { runId },
  })
  .subscribe({
    next: ({ data }) => {
      const ev = data?.onAgentProgress;
      if (!ev) return;
      // paint timeline; fetch artifact from S3 only when DIFF_READY
      setEvents((prev) => [...prev, ev]);
    },
    error: (err) => console.error("subscription", err),
  });

// cleanup on unmount / run complete
// sub.unsubscribe();

#if($context.identity.claims.get("tenant_id") != $context.args.input.tenantId)
  $util.unauthorized()
#end

# ✅ CloudWatch: AppSync 4xx/5xx and connection counts
aws cloudwatch put-metric-alarm \
  --alarm-name appsync-agent-progress-5xx \
  --namespace AWS/AppSync \
  --metric-name 5XXError \
  --dimensions Name=GraphQLAPIId,Value=YOUR_API_ID \
  --statistic Sum --period 60 --threshold 5 \
  --comparison-operator GreaterThanOrEqualToThreshold \
  --evaluation-periods 1