api/runners: accept job logs status and artifacts (S41c)

Status	File	+	-
M	`internal/actions/queries/workflow_jobs.sql`	14	0
M	`internal/actions/queries/workflow_runs.sql`	15	0
M	`internal/actions/queries/workflow_step_log_chunks.sql`	1	0
M	`internal/actions/queries/workflow_steps.sql`	11	0
M	`internal/actions/sqlc/querier.go`	3	0
M	`internal/actions/sqlc/workflow_jobs.sql.go`	57	0
M	`internal/actions/sqlc/workflow_runs.sql.go`	52	0
M	`internal/actions/sqlc/workflow_step_log_chunks.sql.go`	1	0
M	`internal/actions/sqlc/workflow_steps.sql.go`	41	0
M	`internal/web/handlers/api/runners.go`	516	0
M	`internal/web/handlers/api/runners_test.go`	119	10

internal/actions/queries/workflow_jobs.sqlmodified

  FROM workflow_jobs
  WHERE id = $1;
++-- name: UpdateWorkflowJobStatus :one
++UPDATE workflow_jobs
++SET status = $2,
++    conclusion = sqlc.narg(conclusion)::check_conclusion,
++    started_at = sqlc.narg(started_at)::timestamptz,
++    completed_at = sqlc.narg(completed_at)::timestamptz,
++    version = version + 1,
++    updated_at = now()
++WHERE id = $1
++RETURNING id, run_id, job_index, job_key, job_name, runs_on,
++          runner_id, needs_jobs, if_expr, timeout_minutes, permissions,
++          job_env, status, conclusion, cancel_requested,
++          started_at, completed_at, version, created_at, updated_at;
++
  -- name: CountRunningJobsForRunner :one
  SELECT COUNT(*)::integer
  FROM workflow_jobs

internal/actions/queries/workflow_runs.sqlmodified

      updated_at = now()
  WHERE id = $1 AND status = 'queued';
++-- name: CompleteWorkflowRun :one
++UPDATE workflow_runs
++SET status = 'completed',
++    conclusion = sqlc.arg(conclusion)::check_conclusion,
++    started_at = COALESCE(started_at, now()),
++    completed_at = COALESCE(completed_at, now()),
++    version = version + 1,
++    updated_at = now()
++WHERE id = $1
++RETURNING id, repo_id, run_index, workflow_file, workflow_name,
++          head_sha, head_ref, event, event_payload,
++          actor_user_id, parent_run_id, concurrency_group,
++          status, conclusion, pinned, need_approval, approved_by_user_id,
++          started_at, completed_at, version, created_at, updated_at, trigger_event_id;
++
  -- name: NextRunIndexForRepo :one
  -- Atomic next-index emitter: take the max + 1 for this repo. Pairs
  -- with the (repo_id, run_index) UNIQUE so concurrent inserts that

internal/actions/queries/workflow_step_log_chunks.sqlmodified


 -- name: AppendStepLogChunk :one
 INSERT INTO workflow_step_log_chunks (step_id, seq, chunk)
 VALUES ($1, $2, $3)
+ON CONFLICT (step_id, seq) DO NOTHING
 RETURNING id, step_id, seq, created_at;
 
 -- name: ListStepLogChunks :many

internal/actions/queries/workflow_steps.sqlmodified

  WHERE job_id = $1
  ORDER BY step_index ASC;
++-- name: GetFirstStepForJob :one
++SELECT id, job_id, step_index, step_id, step_name, if_expr,
++       run_command, uses_alias, working_directory, step_env,
++       continue_on_error, status, conclusion, log_object_key,
++       log_byte_count, started_at, completed_at, version,
++       created_at, updated_at, step_with
++FROM workflow_steps
++WHERE job_id = $1
++ORDER BY step_index ASC
++LIMIT 1;
++
  -- name: ListStepsForJob :many
  SELECT id, job_id, step_index, step_id, step_name, run_command,
         uses_alias, status, conclusion, log_byte_count,

internal/actions/sqlc/querier.gomodified

  	// SPDX-License-Identifier: AGPL-3.0-or-later
  	AppendStepLogChunk(ctx context.Context, db DBTX, arg AppendStepLogChunkParams) (AppendStepLogChunkRow, error)
  	ClaimQueuedWorkflowJob(ctx context.Context, db DBTX, arg ClaimQueuedWorkflowJobParams) (ClaimQueuedWorkflowJobRow, error)
++	CompleteWorkflowRun(ctx context.Context, db DBTX, arg CompleteWorkflowRunParams) (WorkflowRun, error)
  	CountRunningJobsForRunner(ctx context.Context, db DBTX, runnerID int64) (int32, error)
  	DeleteExpiredArtifacts(ctx context.Context, db DBTX) ([]DeleteExpiredArtifactsRow, error)
  	DeleteExpiredRunnerJWTUses(ctx context.Context, db DBTX) error
  	// target.
  	EnqueueWorkflowRun(ctx context.Context, db DBTX, arg EnqueueWorkflowRunParams) (WorkflowRun, error)
  	GetArtifactByID(ctx context.Context, db DBTX, id int64) (WorkflowArtifact, error)
++	GetFirstStepForJob(ctx context.Context, db DBTX, jobID int64) (WorkflowStep, error)
  	GetOrgSecret(ctx context.Context, db DBTX, arg GetOrgSecretParams) (GetOrgSecretRow, error)
  	GetOrgVariable(ctx context.Context, db DBTX, arg GetOrgVariableParams) (GetOrgVariableRow, error)
  	GetRepoSecret(ctx context.Context, db DBTX, arg GetRepoSecretParams) (GetRepoSecretRow, error)
  	NextRunIndexForRepo(ctx context.Context, db DBTX, repoID int64) (int64, error)
  	RevokeAllTokensForRunner(ctx context.Context, db DBTX, runnerID int64) error
  	TouchRunnerHeartbeat(ctx context.Context, db DBTX, arg TouchRunnerHeartbeatParams) error
++	UpdateWorkflowJobStatus(ctx context.Context, db DBTX, arg UpdateWorkflowJobStatusParams) (WorkflowJob, error)
  	UpsertOrgSecret(ctx context.Context, db DBTX, arg UpsertOrgSecretParams) (WorkflowSecret, error)
  	UpsertOrgVariable(ctx context.Context, db DBTX, arg UpsertOrgVariableParams) (ActionsVariable, error)
  	// SPDX-License-Identifier: AGPL-3.0-or-later

internal/actions/sqlc/workflow_jobs.sql.gomodified

+ 	}
  	return items, nil
+ }
++
++const updateWorkflowJobStatus = `-- name: UpdateWorkflowJobStatus :one
++UPDATE workflow_jobs
++SET status = $2,
++    conclusion = $3::check_conclusion,
++    started_at = $4::timestamptz,
++    completed_at = $5::timestamptz,
++    version = version + 1,
++    updated_at = now()
++WHERE id = $1
++RETURNING id, run_id, job_index, job_key, job_name, runs_on,
++          runner_id, needs_jobs, if_expr, timeout_minutes, permissions,
++          job_env, status, conclusion, cancel_requested,
++          started_at, completed_at, version, created_at, updated_at
++`
++
++type UpdateWorkflowJobStatusParams struct {
++	ID          int64
++	Status      WorkflowJobStatus
++	Conclusion  NullCheckConclusion
++	StartedAt   pgtype.Timestamptz
++	CompletedAt pgtype.Timestamptz
++}
++
++func (q *Queries) UpdateWorkflowJobStatus(ctx context.Context, db DBTX, arg UpdateWorkflowJobStatusParams) (WorkflowJob, error) {
++	row := db.QueryRow(ctx, updateWorkflowJobStatus,
++		arg.ID,
++		arg.Status,
++		arg.Conclusion,
++		arg.StartedAt,
++		arg.CompletedAt,
++	)
++	var i WorkflowJob
++	err := row.Scan(
++		&i.ID,
++		&i.RunID,
++		&i.JobIndex,
++		&i.JobKey,
++		&i.JobName,
++		&i.RunsOn,
++		&i.RunnerID,
++		&i.NeedsJobs,
++		&i.IfExpr,
++		&i.TimeoutMinutes,
++		&i.Permissions,
++		&i.JobEnv,
++		&i.Status,
++		&i.Conclusion,
++		&i.CancelRequested,
++		&i.StartedAt,
++		&i.CompletedAt,
++		&i.Version,
++		&i.CreatedAt,
++		&i.UpdatedAt,
++	)
++	return i, err
++}

internal/actions/sqlc/workflow_runs.sql.gomodified

  	"github.com/jackc/pgx/v5/pgtype"
+ )
++const completeWorkflowRun = `-- name: CompleteWorkflowRun :one
++UPDATE workflow_runs
++SET status = 'completed',
++    conclusion = $2::check_conclusion,
++    started_at = COALESCE(started_at, now()),
++    completed_at = COALESCE(completed_at, now()),
++    version = version + 1,
++    updated_at = now()
++WHERE id = $1
++RETURNING id, repo_id, run_index, workflow_file, workflow_name,
++          head_sha, head_ref, event, event_payload,
++          actor_user_id, parent_run_id, concurrency_group,
++          status, conclusion, pinned, need_approval, approved_by_user_id,
++          started_at, completed_at, version, created_at, updated_at, trigger_event_id
++`
++
++type CompleteWorkflowRunParams struct {
++	ID         int64
++	Conclusion CheckConclusion
++}
++
++func (q *Queries) CompleteWorkflowRun(ctx context.Context, db DBTX, arg CompleteWorkflowRunParams) (WorkflowRun, error) {
++	row := db.QueryRow(ctx, completeWorkflowRun, arg.ID, arg.Conclusion)
++	var i WorkflowRun
++	err := row.Scan(
++		&i.ID,
++		&i.RepoID,
++		&i.RunIndex,
++		&i.WorkflowFile,
++		&i.WorkflowName,
++		&i.HeadSha,
++		&i.HeadRef,
++		&i.Event,
++		&i.EventPayload,
++		&i.ActorUserID,
++		&i.ParentRunID,
++		&i.ConcurrencyGroup,
++		&i.Status,
++		&i.Conclusion,
++		&i.Pinned,
++		&i.NeedApproval,
++		&i.ApprovedByUserID,
++		&i.StartedAt,
++		&i.CompletedAt,
++		&i.Version,
++		&i.CreatedAt,
++		&i.UpdatedAt,
++		&i.TriggerEventID,
++	)
++	return i, err
++}
++
  const enqueueWorkflowRun = `-- name: EnqueueWorkflowRun :one
  INSERT INTO workflow_runs (
      repo_id, run_index, workflow_file, workflow_name,

internal/actions/sqlc/workflow_step_log_chunks.sql.gomodified


 
 INSERT INTO workflow_step_log_chunks (step_id, seq, chunk)
 VALUES ($1, $2, $3)
+ON CONFLICT (step_id, seq) DO NOTHING
 RETURNING id, step_id, seq, created_at
 `
 

internal/actions/sqlc/workflow_steps.sql.gomodified

  	"github.com/jackc/pgx/v5/pgtype"
+ )
++const getFirstStepForJob = `-- name: GetFirstStepForJob :one
++SELECT id, job_id, step_index, step_id, step_name, if_expr,
++       run_command, uses_alias, working_directory, step_env,
++       continue_on_error, status, conclusion, log_object_key,
++       log_byte_count, started_at, completed_at, version,
++       created_at, updated_at, step_with
++FROM workflow_steps
++WHERE job_id = $1
++ORDER BY step_index ASC
++LIMIT 1
++`
++
++func (q *Queries) GetFirstStepForJob(ctx context.Context, db DBTX, jobID int64) (WorkflowStep, error) {
++	row := db.QueryRow(ctx, getFirstStepForJob, jobID)
++	var i WorkflowStep
++	err := row.Scan(
++		&i.ID,
++		&i.JobID,
++		&i.StepIndex,
++		&i.StepID,
++		&i.StepName,
++		&i.IfExpr,
++		&i.RunCommand,
++		&i.UsesAlias,
++		&i.WorkingDirectory,
++		&i.StepEnv,
++		&i.ContinueOnError,
++		&i.Status,
++		&i.Conclusion,
++		&i.LogObjectKey,
++		&i.LogByteCount,
++		&i.StartedAt,
++		&i.CompletedAt,
++		&i.Version,
++		&i.CreatedAt,
++		&i.UpdatedAt,
++		&i.StepWith,
++	)
++	return i, err
++}
++
  const getWorkflowStepByID = `-- name: GetWorkflowStepByID :one
  SELECT id, job_id, step_index, step_id, step_name, if_expr,
         run_command, uses_alias, working_directory, step_env,

internal/web/handlers/api/runners.gomodified

  import (
  	"context"
++	"encoding/base64"
  	"encoding/json"
  	"errors"
  	"fmt"
  	"io"
  	"net/http"
++	"regexp"
++	"strconv"
  	"strings"
  	"time"
  	"github.com/go-chi/chi/v5"
  	"github.com/jackc/pgx/v5"
++	"github.com/jackc/pgx/v5/pgtype"
  	"github.com/tenseleyFlow/shithub/internal/actions/runnerlabels"
  	"github.com/tenseleyFlow/shithub/internal/actions/runnertoken"
  	actionsdb "github.com/tenseleyFlow/shithub/internal/actions/sqlc"
  	"github.com/tenseleyFlow/shithub/internal/auth/runnerjwt"
++	"github.com/tenseleyFlow/shithub/internal/checks"
++	checksdb "github.com/tenseleyFlow/shithub/internal/checks/sqlc"
  	"github.com/tenseleyFlow/shithub/internal/ratelimit"
+ )
  func (h *Handlers) mountRunners(r chi.Router) {
  	r.Post("/api/v1/runners/heartbeat", h.runnerHeartbeat)
++	r.Post("/api/v1/jobs/{id}/logs", h.runnerJobLogs)
++	r.Post("/api/v1/jobs/{id}/status", h.runnerJobStatus)
++	r.Post("/api/v1/jobs/{id}/artifacts/upload", h.runnerJobArtifactUpload)
++	r.Post("/api/v1/jobs/{id}/cancel-check", h.runnerJobCancelCheck)
+ }
  type runnerHeartbeatRequest struct {
  	return job, steps, true, nil
+ }
++type runnerJobAuth struct {
++	Claims   runnerjwt.Claims
++	RunnerID int64
++	Job      actionsdb.WorkflowJob
++}
++
++func (h *Handlers) authenticateRunnerJob(w http.ResponseWriter, r *http.Request) (runnerJobAuth, bool) {
++	if h.d.RunnerJWT == nil {
++		writeAPIError(w, http.StatusServiceUnavailable, "runner API is not configured")
++		return runnerJobAuth{}, false
++	}
++	pathJobID, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
++	if err != nil || pathJobID <= 0 {
++		writeAPIError(w, http.StatusNotFound, "job not found")
++		return runnerJobAuth{}, false
++	}
++	const prefix = "Bearer "
++	authz := r.Header.Get("Authorization")
++	if !strings.HasPrefix(authz, prefix) {
++		writeAPIError(w, http.StatusUnauthorized, "job token required")
++		return runnerJobAuth{}, false
++	}
++	claims, err := h.d.RunnerJWT.Verify(strings.TrimSpace(strings.TrimPrefix(authz, prefix)))
++	if err != nil {
++		writeAPIError(w, http.StatusUnauthorized, "job token invalid")
++		return runnerJobAuth{}, false
++	}
++	if claims.JobID != pathJobID {
++		writeAPIError(w, http.StatusNotFound, "job not found")
++		return runnerJobAuth{}, false
++	}
++	runnerID, err := claims.RunnerID()
++	if err != nil {
++		writeAPIError(w, http.StatusUnauthorized, "job token invalid")
++		return runnerJobAuth{}, false
++	}
++	job, err := actionsdb.New().GetWorkflowJobByID(r.Context(), h.d.Pool, pathJobID)
++	if err != nil {
++		if errors.Is(err, pgx.ErrNoRows) {
++			writeAPIError(w, http.StatusNotFound, "job not found")
++		} else {
++			writeAPIError(w, http.StatusInternalServerError, "job lookup failed")
++		}
++		return runnerJobAuth{}, false
++	}
++	if job.RunID != claims.RunID || !job.RunnerID.Valid || job.RunnerID.Int64 != runnerID {
++		writeAPIError(w, http.StatusNotFound, "job not found")
++		return runnerJobAuth{}, false
++	}
++	if err := runnerjwt.Consume(r.Context(), h.d.Pool, claims); err != nil {
++		if errors.Is(err, runnerjwt.ErrReplay) {
++			writeAPIError(w, http.StatusUnauthorized, "job token replayed")
++		} else {
++			h.d.Logger.ErrorContext(r.Context(), "runner jwt consume failed", "job_id", pathJobID, "error", err)
++			writeAPIError(w, http.StatusUnauthorized, "job token invalid")
++		}
++		return runnerJobAuth{}, false
++	}
++	return runnerJobAuth{Claims: claims, RunnerID: runnerID, Job: job}, true
++}
++
++type runnerLogRequest struct {
++	Seq    int32  `json:"seq"`
++	Chunk  string `json:"chunk"`
++	StepID int64  `json:"step_id,omitempty"`
++}
++
++func (h *Handlers) runnerJobLogs(w http.ResponseWriter, r *http.Request) {
++	auth, ok := h.authenticateRunnerJob(w, r)
++	if !ok {
++		return
++	}
++	var body runnerLogRequest
++	if err := decodeJSONBody(r.Body, &body); err != nil {
++		writeAPIError(w, http.StatusBadRequest, "invalid JSON: "+err.Error())
++		return
++	}
++	if body.Seq < 0 {
++		writeAPIError(w, http.StatusBadRequest, "seq must be non-negative")
++		return
++	}
++	chunk, err := decodeBase64(body.Chunk)
++	if err != nil {
++		writeAPIError(w, http.StatusBadRequest, "chunk must be base64")
++		return
++	}
++	if len(chunk) == 0 || len(chunk) > 512*1024 {
++		writeAPIError(w, http.StatusBadRequest, "chunk must be between 1 and 524288 bytes")
++		return
++	}
++	stepID, ok := h.resolveLogStep(w, r, auth.Job.ID, body.StepID)
++	if !ok {
++		return
++	}
++	if _, err := actionsdb.New().AppendStepLogChunk(r.Context(), h.d.Pool, actionsdb.AppendStepLogChunkParams{
++		StepID: stepID,
++		Seq:    body.Seq,
++		Chunk:  chunk,
++	}); err != nil && !errors.Is(err, pgx.ErrNoRows) {
++		writeAPIError(w, http.StatusInternalServerError, "append log failed")
++		return
++	}
++	h.writeNextTokenResponse(w, r, http.StatusAccepted, auth, map[string]any{"accepted": true})
++}
++
++func (h *Handlers) resolveLogStep(w http.ResponseWriter, r *http.Request, jobID, stepID int64) (int64, bool) {
++	q := actionsdb.New()
++	if stepID == 0 {
++		step, err := q.GetFirstStepForJob(r.Context(), h.d.Pool, jobID)
++		if err != nil {
++			writeAPIError(w, http.StatusNotFound, "step not found")
++			return 0, false
++		}
++		return step.ID, true
++	}
++	step, err := q.GetWorkflowStepByID(r.Context(), h.d.Pool, stepID)
++	if err != nil || step.JobID != jobID {
++		writeAPIError(w, http.StatusNotFound, "step not found")
++		return 0, false
++	}
++	return step.ID, true
++}
++
++type runnerStatusRequest struct {
++	Status      string `json:"status"`
++	Conclusion  string `json:"conclusion,omitempty"`
++	StartedAt   string `json:"started_at,omitempty"`
++	CompletedAt string `json:"completed_at,omitempty"`
++}
++
++func (h *Handlers) runnerJobStatus(w http.ResponseWriter, r *http.Request) {
++	auth, ok := h.authenticateRunnerJob(w, r)
++	if !ok {
++		return
++	}
++	var body runnerStatusRequest
++	if err := decodeJSONBody(r.Body, &body); err != nil {
++		writeAPIError(w, http.StatusBadRequest, "invalid JSON: "+err.Error())
++		return
++	}
++	update, terminal, err := normalizeJobStatusUpdate(auth.Job, body)
++	if err != nil {
++		writeAPIError(w, http.StatusBadRequest, err.Error())
++		return
++	}
++	updated, runCompleted, runConclusion, err := h.applyJobStatus(r.Context(), auth.Job, update)
++	if err != nil {
++		writeAPIError(w, http.StatusInternalServerError, "status update failed")
++		return
++	}
++	if err := h.updateCheckRunForJob(r.Context(), updated); err != nil {
++		h.d.Logger.WarnContext(r.Context(), "runner check_run update failed", "job_id", updated.ID, "error", err)
++	}
++
++	bodyMap := map[string]any{
++		"status":     string(updated.Status),
++		"conclusion": nullableConclusion(updated.Conclusion),
++	}
++	if runCompleted {
++		bodyMap["run_status"] = "completed"
++		bodyMap["run_conclusion"] = string(runConclusion)
++	}
++	if terminal {
++		writeJSON(w, http.StatusOK, bodyMap)
++		return
++	}
++	h.writeNextTokenResponse(w, r, http.StatusOK, auth, bodyMap)
++}
++
++type normalizedJobStatusUpdate struct {
++	Status      actionsdb.WorkflowJobStatus
++	Conclusion  actionsdb.NullCheckConclusion
++	StartedAt   pgtype.Timestamptz
++	CompletedAt pgtype.Timestamptz
++}
++
++func normalizeJobStatusUpdate(job actionsdb.WorkflowJob, body runnerStatusRequest) (normalizedJobStatusUpdate, bool, error) {
++	now := time.Now().UTC()
++	status := actionsdb.WorkflowJobStatus(strings.TrimSpace(body.Status))
++	if status == "" {
++		return normalizedJobStatusUpdate{}, false, errors.New("status is required")
++	}
++	if !validWorkflowJobTransition(job.Status, status) {
++		return normalizedJobStatusUpdate{}, false, fmt.Errorf("invalid status transition %s -> %s", job.Status, status)
++	}
++	startedAt := job.StartedAt
++	if body.StartedAt != "" {
++		t, err := parseTimeOptional(body.StartedAt)
++		if err != nil {
++			return normalizedJobStatusUpdate{}, false, fmt.Errorf("started_at: %w", err)
++		}
++		startedAt = pgtype.Timestamptz{Time: t, Valid: !t.IsZero()}
++	}
++	if !startedAt.Valid && (status == actionsdb.WorkflowJobStatusRunning ||
++		status == actionsdb.WorkflowJobStatusCompleted ||
++		status == actionsdb.WorkflowJobStatusCancelled) {
++		startedAt = pgtype.Timestamptz{Time: now, Valid: true}
++	}
++	completedAt := job.CompletedAt
++	terminal := status == actionsdb.WorkflowJobStatusCompleted || status == actionsdb.WorkflowJobStatusCancelled
++	if body.CompletedAt != "" {
++		t, err := parseTimeOptional(body.CompletedAt)
++		if err != nil {
++			return normalizedJobStatusUpdate{}, false, fmt.Errorf("completed_at: %w", err)
++		}
++		completedAt = pgtype.Timestamptz{Time: t, Valid: !t.IsZero()}
++	}
++	if terminal && !completedAt.Valid {
++		completedAt = pgtype.Timestamptz{Time: now, Valid: true}
++	}
++	conclusion := actionsdb.NullCheckConclusion{}
++	if terminal {
++		c := strings.TrimSpace(body.Conclusion)
++		if c == "" && status == actionsdb.WorkflowJobStatusCancelled {
++			c = "cancelled"
++		}
++		if !validRunnerConclusion(c) {
++			return normalizedJobStatusUpdate{}, false, errors.New("invalid or missing conclusion")
++		}
++		conclusion = actionsdb.NullCheckConclusion{CheckConclusion: actionsdb.CheckConclusion(c), Valid: true}
++	} else if strings.TrimSpace(body.Conclusion) != "" {
++		return normalizedJobStatusUpdate{}, false, errors.New("conclusion is only valid for terminal statuses")
++	}
++	return normalizedJobStatusUpdate{
++		Status:      status,
++		Conclusion:  conclusion,
++		StartedAt:   startedAt,
++		CompletedAt: completedAt,
++	}, terminal, nil
++}
++
++func validWorkflowJobTransition(from, to actionsdb.WorkflowJobStatus) bool {
++	switch to {
++	case actionsdb.WorkflowJobStatusRunning:
++		return from == actionsdb.WorkflowJobStatusQueued || from == actionsdb.WorkflowJobStatusRunning
++	case actionsdb.WorkflowJobStatusCompleted:
++		return from == actionsdb.WorkflowJobStatusQueued || from == actionsdb.WorkflowJobStatusRunning || from == actionsdb.WorkflowJobStatusCompleted
++	case actionsdb.WorkflowJobStatusCancelled:
++		return from == actionsdb.WorkflowJobStatusQueued || from == actionsdb.WorkflowJobStatusRunning || from == actionsdb.WorkflowJobStatusCancelled
++	default:
++		return false
++	}
++}
++
++func (h *Handlers) applyJobStatus(
++	ctx context.Context,
++	job actionsdb.WorkflowJob,
++	update normalizedJobStatusUpdate,
++) (actionsdb.WorkflowJob, bool, actionsdb.CheckConclusion, error) {
++	q := actionsdb.New()
++	tx, err := h.d.Pool.Begin(ctx)
++	if err != nil {
++		return actionsdb.WorkflowJob{}, false, "", err
++	}
++	committed := false
++	defer func() {
++		if !committed {
++			_ = tx.Rollback(ctx)
++		}
++	}()
++	updated, err := q.UpdateWorkflowJobStatus(ctx, tx, actionsdb.UpdateWorkflowJobStatusParams{
++		ID:          job.ID,
++		Status:      update.Status,
++		Conclusion:  update.Conclusion,
++		StartedAt:   update.StartedAt,
++		CompletedAt: update.CompletedAt,
++	})
++	if err != nil {
++		return actionsdb.WorkflowJob{}, false, "", err
++	}
++	jobs, err := q.ListJobsForRun(ctx, tx, updated.RunID)
++	if err != nil {
++		return actionsdb.WorkflowJob{}, false, "", err
++	}
++	runConclusion, complete := deriveWorkflowRunConclusion(jobs)
++	if complete {
++		if _, err := q.CompleteWorkflowRun(ctx, tx, actionsdb.CompleteWorkflowRunParams{
++			ID:         updated.RunID,
++			Conclusion: runConclusion,
++		}); err != nil {
++			return actionsdb.WorkflowJob{}, false, "", err
++		}
++	} else if err := q.MarkWorkflowRunRunning(ctx, tx, updated.RunID); err != nil {
++		return actionsdb.WorkflowJob{}, false, "", err
++	}
++	if err := tx.Commit(ctx); err != nil {
++		return actionsdb.WorkflowJob{}, false, "", err
++	}
++	committed = true
++	return updated, complete, runConclusion, nil
++}
++
++func deriveWorkflowRunConclusion(jobs []actionsdb.ListJobsForRunRow) (actionsdb.CheckConclusion, bool) {
++	if len(jobs) == 0 {
++		return actionsdb.CheckConclusionFailure, true
++	}
++	worst := actionsdb.CheckConclusionSuccess
++	for _, job := range jobs {
++		switch job.Status {
++		case actionsdb.WorkflowJobStatusCompleted, actionsdb.WorkflowJobStatusCancelled, actionsdb.WorkflowJobStatusSkipped:
++		default:
++			return "", false
++		}
++		if job.Status == actionsdb.WorkflowJobStatusCancelled {
++			worst = actionsdb.CheckConclusionCancelled
++			continue
++		}
++		if !job.Conclusion.Valid {
++			return actionsdb.CheckConclusionFailure, true
++		}
++		c := job.Conclusion.CheckConclusion
++		if c == actionsdb.CheckConclusionFailure ||
++			c == actionsdb.CheckConclusionTimedOut ||
++			c == actionsdb.CheckConclusionActionRequired {
++			return c, true
++		}
++		if c == actionsdb.CheckConclusionCancelled {
++			worst = actionsdb.CheckConclusionCancelled
++		}
++	}
++	return worst, true
++}
++
++func (h *Handlers) updateCheckRunForJob(ctx context.Context, job actionsdb.WorkflowJob) error {
++	run, err := actionsdb.New().GetWorkflowRunByID(ctx, h.d.Pool, job.RunID)
++	if err != nil {
++		return err
++	}
++	name := job.JobName
++	if name == "" {
++		name = job.JobKey
++	}
++	checkRun, err := checksdb.New().GetCheckRunByExternalID(ctx, h.d.Pool, checksdb.GetCheckRunByExternalIDParams{
++		RepoID:     run.RepoID,
++		HeadSha:    run.HeadSha,
++		Name:       name,
++		ExternalID: pgtype.Text{String: fmt.Sprintf("workflow_run:%d:job:%s", job.RunID, job.JobKey), Valid: true},
++	})
++	if err != nil {
++		return err
++	}
++	params := checks.UpdateParams{
++		RunID:        checkRun.ID,
++		HasStatus:    true,
++		HasStartedAt: true,
++		StartedAt:    timeFromPg(job.StartedAt),
++	}
++	switch job.Status {
++	case actionsdb.WorkflowJobStatusRunning:
++		params.Status = "in_progress"
++	case actionsdb.WorkflowJobStatusCompleted, actionsdb.WorkflowJobStatusCancelled:
++		params.Status = "completed"
++		params.HasConclusion = true
++		if job.Conclusion.Valid {
++			params.Conclusion = string(job.Conclusion.CheckConclusion)
++		} else if job.Status == actionsdb.WorkflowJobStatusCancelled {
++			params.Conclusion = "cancelled"
++		}
++		params.HasCompletedAt = true
++		params.CompletedAt = timeFromPg(job.CompletedAt)
++	default:
++		return nil
++	}
++	_, err = checks.Update(ctx, checks.Deps{Pool: h.d.Pool, Logger: h.d.Logger}, params)
++	return err
++}
++
++type runnerArtifactUploadRequest struct {
++	Name      string `json:"name"`
++	SizeBytes int64  `json:"size_bytes"`
++}
++
++var artifactNameRE = regexp.MustCompile(`^[A-Za-z0-9._-]+$`)
++
++func (h *Handlers) runnerJobArtifactUpload(w http.ResponseWriter, r *http.Request) {
++	if h.d.ObjectStore == nil {
++		writeAPIError(w, http.StatusServiceUnavailable, "object storage is not configured")
++		return
++	}
++	auth, ok := h.authenticateRunnerJob(w, r)
++	if !ok {
++		return
++	}
++	var body runnerArtifactUploadRequest
++	if err := decodeJSONBody(r.Body, &body); err != nil {
++		writeAPIError(w, http.StatusBadRequest, "invalid JSON: "+err.Error())
++		return
++	}
++	body.Name = strings.TrimSpace(body.Name)
++	if !validArtifactName(body.Name) {
++		writeAPIError(w, http.StatusBadRequest, "invalid artifact name")
++		return
++	}
++	if body.SizeBytes < 0 {
++		writeAPIError(w, http.StatusBadRequest, "size_bytes must be non-negative")
++		return
++	}
++	objectKey := fmt.Sprintf("actions/runs/%d/artifacts/%s", auth.Claims.RunID, body.Name)
++	artifact, err := actionsdb.New().InsertArtifact(r.Context(), h.d.Pool, actionsdb.InsertArtifactParams{
++		RunID:     auth.Claims.RunID,
++		Name:      body.Name,
++		ObjectKey: objectKey,
++		ByteCount: body.SizeBytes,
++		ExpiresAt: pgtype.Timestamptz{
++			Time:  time.Now().UTC().Add(90 * 24 * time.Hour),
++			Valid: true,
++		},
++	})
++	if err != nil {
++		writeAPIError(w, http.StatusInternalServerError, "artifact create failed")
++		return
++	}
++	uploadURL, err := h.d.ObjectStore.SignedURL(r.Context(), objectKey, 15*time.Minute, http.MethodPut)
++	if err != nil {
++		writeAPIError(w, http.StatusInternalServerError, "artifact upload url failed")
++		return
++	}
++	h.writeNextTokenResponse(w, r, http.StatusCreated, auth, map[string]any{
++		"artifact_id": artifact.ID,
++		"upload_url":  uploadURL,
++	})
++}
++
++func validArtifactName(name string) bool {
++	return len(name) >= 1 &&
++		len(name) <= 100 &&
++		artifactNameRE.MatchString(name) &&
++		!strings.HasPrefix(name, "..") &&
++		!strings.Contains(name, "/")
++}
++
++func (h *Handlers) runnerJobCancelCheck(w http.ResponseWriter, r *http.Request) {
++	auth, ok := h.authenticateRunnerJob(w, r)
++	if !ok {
++		return
++	}
++	h.writeNextTokenResponse(w, r, http.StatusOK, auth, map[string]any{
++		"cancelled": auth.Job.CancelRequested,
++	})
++}
++
++func (h *Handlers) writeNextTokenResponse(
++	w http.ResponseWriter,
++	r *http.Request,
++	status int,
++	auth runnerJobAuth,
++	body map[string]any,
++) {
++	token, claims, err := h.d.RunnerJWT.Mint(runnerjwt.MintParams{
++		RunnerID: auth.RunnerID,
++		JobID:    auth.Claims.JobID,
++		RunID:    auth.Claims.RunID,
++		RepoID:   auth.Claims.RepoID,
++	})
++	if err != nil {
++		h.d.Logger.ErrorContext(r.Context(), "runner next-token mint failed", "job_id", auth.Claims.JobID, "error", err)
++		writeAPIError(w, http.StatusInternalServerError, "runner token mint failed")
++		return
++	}
++	body["next_token"] = token
++	body["next_token_expires_at"] = time.Unix(claims.Exp, 0).UTC().Format(time.RFC3339)
++	writeJSON(w, status, body)
++}
++
  type runnerClaimResponse struct {
  	Token     string           `json:"token"`
  	ExpiresAt string           `json:"expires_at"`
+ 	}
  	return json.RawMessage(b)
+ }
++
++func decodeJSONBody(r io.Reader, v any) error {
++	dec := json.NewDecoder(r)
++	dec.DisallowUnknownFields()
++	return dec.Decode(v)
++}
++
++func decodeBase64(s string) ([]byte, error) {
++	if out, err := base64.StdEncoding.DecodeString(s); err == nil {
++		return out, nil
++	}
++	return base64.RawStdEncoding.DecodeString(s)
++}
++
++func validRunnerConclusion(c string) bool {
++	switch actionsdb.CheckConclusion(c) {
++	case actionsdb.CheckConclusionSuccess,
++		actionsdb.CheckConclusionFailure,
++		actionsdb.CheckConclusionNeutral,
++		actionsdb.CheckConclusionCancelled,
++		actionsdb.CheckConclusionSkipped,
++		actionsdb.CheckConclusionTimedOut,
++		actionsdb.CheckConclusionActionRequired:
++		return true
++	default:
++		return false
++	}
++}
++
++func nullableConclusion(c actionsdb.NullCheckConclusion) any {
++	if !c.Valid {
++		return nil
++	}
++	return string(c.CheckConclusion)
++}
++
++func timeFromPg(t pgtype.Timestamptz) time.Time {
++	if !t.Valid {
++		return time.Time{}
++	}
++	return t.Time
++}

internal/web/handlers/api/runners_test.gomodified

  import (
  	"bytes"
  	"context"
++	"encoding/base64"
  	"encoding/json"
++	"fmt"
  	"io"
  	"log/slog"
  	"net/http"
  	"github.com/tenseleyFlow/shithub/internal/actions/trigger"
  	"github.com/tenseleyFlow/shithub/internal/actions/workflow"
  	"github.com/tenseleyFlow/shithub/internal/auth/runnerjwt"
++	"github.com/tenseleyFlow/shithub/internal/infra/storage"
  	reposdb "github.com/tenseleyFlow/shithub/internal/repos/sqlc"
  	"github.com/tenseleyFlow/shithub/internal/testing/dbtest"
  	usersdb "github.com/tenseleyFlow/shithub/internal/users/sqlc"
  		t.Fatalf("claims runner_id: got %d, want %d", claimRunnerID, runnerID)
+ 	}
++	var logResp struct {
++		Accepted  bool   `json:"accepted"`
++		NextToken string `json:"next_token"`
++	}
++	logBody := fmt.Sprintf(`{"seq":0,"chunk":%q}`, base64.StdEncoding.EncodeToString([]byte("hello\n")))
++	req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/jobs/%d/logs", resp.Job.ID), strings.NewReader(logBody))
++	req.Header.Set("Authorization", "Bearer "+resp.Token)
++	rr = httptest.NewRecorder()
++	router.ServeHTTP(rr, req)
++	if rr.Code != http.StatusAccepted {
++		t.Fatalf("logs status: got %d, want 202; body=%s", rr.Code, rr.Body.String())
++	}
++	if err := json.Unmarshal(rr.Body.Bytes(), &logResp); err != nil {
++		t.Fatalf("decode log response: %v", err)
++	}
++	if !logResp.Accepted || logResp.NextToken == "" {
++		t.Fatalf("unexpected log response: %+v", logResp)
++	}
++
++	req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/jobs/%d/logs", resp.Job.ID), strings.NewReader(logBody))
++	req.Header.Set("Authorization", "Bearer "+resp.Token)
++	rr = httptest.NewRecorder()
++	router.ServeHTTP(rr, req)
++	if rr.Code != http.StatusUnauthorized {
++		t.Fatalf("replay status: got %d, want 401; body=%s", rr.Code, rr.Body.String())
++	}
++
++	req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/jobs/%d/status", resp.Job.ID),
++		strings.NewReader(`{"status":"completed","conclusion":"success"}`))
++	req.Header.Set("Authorization", "Bearer "+logResp.NextToken)
++	rr = httptest.NewRecorder()
++	router.ServeHTTP(rr, req)
++	if rr.Code != http.StatusOK {
++		t.Fatalf("complete status: got %d, want 200; body=%s", rr.Code, rr.Body.String())
++	}
++
  	q := actionsdb.New()
  	job, err := q.GetWorkflowJobByID(ctx, pool, resp.Job.ID)
  	if err != nil {
  		t.Fatalf("GetWorkflowJobByID: %v", err)
+ 	}
--	if job.Status != actionsdb.WorkflowJobStatusRunning || !job.RunnerID.Valid || job.RunnerID.Int64 != runnerID {
++	if job.Status != actionsdb.WorkflowJobStatusCompleted || !job.RunnerID.Valid || job.RunnerID.Int64 != runnerID ||
--		t.Fatalf("job not claimed by runner: %+v", job)
++		!job.Conclusion.Valid || job.Conclusion.CheckConclusion != actionsdb.CheckConclusionSuccess {
++		t.Fatalf("job not completed by runner: %+v", job)
+ 	}
  	run, err := q.GetWorkflowRunByID(ctx, pool, runID)
  	if err != nil {
  		t.Fatalf("GetWorkflowRunByID: %v", err)
+ 	}
--	if run.Status != actionsdb.WorkflowRunStatusRunning {
++	if run.Status != actionsdb.WorkflowRunStatusCompleted ||
--		t.Fatalf("run status: got %s, want running", run.Status)
++		!run.Conclusion.Valid || run.Conclusion.CheckConclusion != actionsdb.CheckConclusionSuccess {
++		t.Fatalf("run not completed successfully: %+v", run)
+ 	}
--	// Capacity is enforced server-side: a second heartbeat from the same
++	// The completed job is no longer counted against capacity, and no other
--	// runner sees one running job and receives no additional claim.
++	// queued job exists, so the heartbeat is an empty 204.
  	req = httptest.NewRequest(http.MethodPost, "/api/v1/runners/heartbeat",
  		strings.NewReader(`{"labels":["ubuntu-latest","linux"],"capacity":1}`))
  	req.Header.Set("Authorization", "Bearer "+token)
+ 	}
+ }
--func newRunnerAPIRouter(t *testing.T, pool *pgxpool.Pool, logger *slog.Logger, signer *runnerjwt.Signer) http.Handler {
++func TestRunnerArtifactUploadReturnsSignedURL(t *testing.T) {
++	ctx := context.Background()
++	pool := dbtest.NewTestDB(t)
++	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
++	repoID, userID := setupRunnerAPIRepo(t, pool)
++	enqueueRunnerAPIRun(t, pool, logger, repoID, userID)
++	token, _ := registerRunnerForTest(t, pool, []string{"ubuntu-latest"}, 1)
++	router := newRunnerAPIRouter(t, pool, logger, runnerAPISigner(t, time.Now()), storage.NewMemoryStore())
++
++	req := httptest.NewRequest(http.MethodPost, "/api/v1/runners/heartbeat",
++		strings.NewReader(`{"labels":["ubuntu-latest"],"capacity":1}`))
++	req.Header.Set("Authorization", "Bearer "+token)
++	rr := httptest.NewRecorder()
++	router.ServeHTTP(rr, req)
++	if rr.Code != http.StatusOK {
++		t.Fatalf("heartbeat status: got %d, want 200; body=%s", rr.Code, rr.Body.String())
++	}
++	var claim struct {
++		Token string `json:"token"`
++		Job   struct {
++			ID    int64 `json:"id"`
++			RunID int64 `json:"run_id"`
++		} `json:"job"`
++	}
++	if err := json.Unmarshal(rr.Body.Bytes(), &claim); err != nil {
++		t.Fatalf("decode claim: %v", err)
++	}
++
++	req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/jobs/%d/artifacts/upload", claim.Job.ID),
++		strings.NewReader(`{"name":"test-results.tgz","size_bytes":123}`))
++	req.Header.Set("Authorization", "Bearer "+claim.Token)
++	rr = httptest.NewRecorder()
++	router.ServeHTTP(rr, req)
++	if rr.Code != http.StatusCreated {
++		t.Fatalf("artifact status: got %d, want 201; body=%s", rr.Code, rr.Body.String())
++	}
++	var upload struct {
++		ArtifactID int64  `json:"artifact_id"`
++		UploadURL  string `json:"upload_url"`
++		NextToken  string `json:"next_token"`
++	}
++	if err := json.Unmarshal(rr.Body.Bytes(), &upload); err != nil {
++		t.Fatalf("decode upload: %v", err)
++	}
++	if upload.ArtifactID == 0 || upload.NextToken == "" ||
++		!strings.HasPrefix(upload.UploadURL, "mem://actions/runs/") {
++		t.Fatalf("unexpected upload response: %+v", upload)
++	}
++	artifacts, err := actionsdb.New().ListArtifactsForRun(ctx, pool, claim.Job.RunID)
++	if err != nil {
++		t.Fatalf("ListArtifactsForRun: %v", err)
++	}
++	if len(artifacts) != 1 || artifacts[0].Name != "test-results.tgz" || artifacts[0].ByteCount != 123 {
++		t.Fatalf("unexpected artifacts: %+v", artifacts)
++	}
++}
++
++func newRunnerAPIRouter(
++	t *testing.T,
++	pool *pgxpool.Pool,
++	logger *slog.Logger,
++	signer *runnerjwt.Signer,
++	stores ...storage.ObjectStore,
++) http.Handler {
  	t.Helper()
++	var store storage.ObjectStore
++	if len(stores) > 0 {
++		store = stores[0]
++	}
  	h, err := apih.New(apih.Deps{
--		Pool:      pool,
++		Pool:        pool,
--		Logger:    logger,
++		Logger:      logger,
--		RunnerJWT: signer,
++		RunnerJWT:   signer,
++		ObjectStore: store,
  	})
  	if err != nil {
  		t.Fatalf("api.New: %v", err)

tenseleyflow/shithub / `b70f1f2`

11 changed files

`@@ -3,6 +3,7 @@`
3	-- name: AppendStepLogChunk :one	3	-- name: AppendStepLogChunk :one
4	INSERT INTO workflow_step_log_chunks (step_id, seq, chunk)	4	INSERT INTO workflow_step_log_chunks (step_id, seq, chunk)
5	VALUES ($1, $2, $3)	5	VALUES ($1, $2, $3)
		6	+ON CONFLICT (step_id, seq) DO NOTHING
6	RETURNING id, step_id, seq, created_at;	7	RETURNING id, step_id, seq, created_at;
7		8
8	-- name: ListStepLogChunks :many	9	-- name: ListStepLogChunks :many

`@@ -15,6 +15,7 @@` const appendStepLogChunk = `-- name: AppendStepLogChunk :one
15		15
16	INSERT INTO workflow_step_log_chunks (step_id, seq, chunk)	16	INSERT INTO workflow_step_log_chunks (step_id, seq, chunk)
17	VALUES ($1, $2, $3)	17	VALUES ($1, $2, $3)
		18	+ON CONFLICT (step_id, seq) DO NOTHING
18	RETURNING id, step_id, seq, created_at	19	RETURNING id, step_id, seq, created_at
19	`	20	`
20		21