Tasks

Integrate an agent

Publish a taskable agent, process assigned tasks, and report status using the Lattice SDK

A taskable agent is an asset, or a group of assets, that listens for tasks assigned to it in Lattice, executes them, and reports its progress back. This guide walks through the full agent workflow: publishing a taskable agent, streaming and parsing assigned tasks, updating task status, and handling cancellation requests.

An agent calls the following API to listen for tasks assigned to it in Lattice:

  • StreamAsAgent — For monitoring tasks routed to the agent using REST.
  • ListenAsAgent — For monitoring tasks routed to the agent using gRPC.

Before you begin

  • To publish taskable entities, and subscribe to tasks, set up your Lattice environment.
  • Familiarize yourself with entities and different entity types.
  • Review how to define a task. This guide uses the Objective task definition from that page as its example.
gRPC authentication

These gRPC examples authenticate with a static environment token attached as request metadata. If you are using OAuth 2.0 client credentials instead, set up the token refresh module before running the examples on this page.

Publish a taskable agent

An asset is an entity under your control, or under the control of another operator or system. Assets may accept tasks such as search or tracking. An agent is an asset, or a group of assets, that can complete a specific set of defined tasks.

To publish an agent, do the following:

1

Define a TaskCatalog

The entity model’s TaskCatalog component defines the tasks that an asset can execute.

For example, if your asset supports the Objective task, publish an asset entity that includes it in its catalog, and listen for and execute that task. The taskSpecificationUrl is the type URL of the task definition you authored and published to the Schema Registry:

taskCatalog
1  "taskCatalog": {
2      // Define the tasks the asset can perform.
3      "taskDefinitions": [
4          {
5              // Set the task specification URL to your Objective definition.
6              "taskSpecificationUrl": "type.googleapis.com/org.example.reconnaissance.v1beta.Objective",
7          }
8      ]
9  }
2

Publish the agent

Use the PublishEntity method to publish a taskable agent:

1package main
2
3import (
4	"context"
5	"fmt"
6	"net/http"
7	"os"
8	"time"
9
10	Lattice "github.com/anduril/lattice-sdk-go/v5"
11	"github.com/anduril/lattice-sdk-go/v5/client"
12	"github.com/anduril/lattice-sdk-go/v5/option"
13	"github.com/google/uuid"
14)
15
16func main() {
17	// Get environment variables
18	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
19	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
20
21	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes
22	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
23
24	// Check required environment variables
25	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
26		fmt.Println("Missing required environment variables")
27		os.Exit(1)
28	}
29
30	// Initialize headers for sandbox authorization
31	headers := http.Header{}
32	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
33
34	// Create the client
35	LatticeClient := client.NewClient(
36		option.WithToken(environmentToken),
37		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
38		option.WithHTTPHeader(headers),
39	)
40
41	// Generate a unique ID for the entity
42	entityId := uuid.New().String()
43
44	// Get creation time
45	creationTime := time.Now().UTC()
46
47	// Continuously publish the entity
48	for {
49		latestTimestamp := time.Now().UTC()
50		ctx := context.Background()
51
52		// Create entity to publish
53		entity := Lattice.Entity{
54			EntityID:    &entityId,
55			Description: Lattice.String("Friendly drone asset"),
56			Aliases: &Lattice.Aliases{
57				Name: Lattice.String("Drone 1"),
58			},
59			IsLive:      Lattice.Bool(true),
60			CreatedTime: Lattice.Time(creationTime),
61			ExpiryTime:  Lattice.Time(latestTimestamp.Add(1 * time.Minute)),
62			Ontology: &Lattice.Ontology{
63				Template:     Lattice.OntologyTemplateTemplateAsset.Ptr(),
64				PlatformType: Lattice.String("UAV"),
65			},
66			MilView: &Lattice.MilView{
67				Disposition: Lattice.MilViewDispositionDispositionFriendly.Ptr(),
68				Environment: Lattice.MilViewEnvironmentEnvironmentAir.Ptr(),
69			},
70			Location: &Lattice.Location{
71				Position: &Lattice.Position{
72					LatitudeDegrees:   Lattice.Float64(50.91402185768586),
73					LongitudeDegrees:  Lattice.Float64(0.79203612077257),
74					AltitudeAsfMeters: Lattice.Float64(1000),
75				},
76			},
77			Provenance: &Lattice.Provenance{
78				IntegrationName:  Lattice.String("your_integration_name"),
79				DataType:         Lattice.String("your_data_type"),
80				SourceUpdateTime: Lattice.Time(latestTimestamp),
81			},
82			Health: &Lattice.Health{
83				ConnectionStatus: Lattice.HealthConnectionStatusConnectionStatusOnline.Ptr(),
84				HealthStatus:     Lattice.HealthHealthStatusHealthStatusHealthy.Ptr(),
85				UpdateTime:       Lattice.Time(latestTimestamp),
86			},
87			TaskCatalog: &Lattice.TaskCatalog{
88				TaskDefinitions: []*Lattice.TaskDefinition{
89					{TaskSpecificationURL: Lattice.String("type.googleapis.com/anduril.tasks.v2.VisualId")},
90					{TaskSpecificationURL: Lattice.String("type.googleapis.com/anduril.tasks.v2.Investigate")},
91				},
92			},
93		}
94
95		// Publish the entity
96		_, err := LatticeClient.Entities.PublishEntity(ctx, &entity)
97
98		// Handle errors
99		if err != nil {
100			fmt.Printf("Error publishing entity: %v\n", err)
101		} else {
102			fmt.Println("Published asset: " + entityId)
103		}
104
105		// Wait before next request
106		time.Sleep(5 * time.Second)
107	}
108}

If successful, you see the entity ID in the console:

$2025-07-16T02:50:04.694Z [INFO]: Published asset: <entity-id>

Process assigned tasks

Once an agent is published, the integration should open a stream to receive tasks routed to it. Each task carries a specification — a google.protobuf.Any that holds the task’s data. To act on a task, the agent reads the metadata off the request, then decodes the specification into the task type it advertised in its catalog.

1

Stream tasks as an agent

Use StreamAsAgent, or ListenAsAgent for gRPC, to open the stream. When an executeRequest arrives, the agent reads the task ID and status version, parses the task data, then reports that it has started the task.

Task delivery over a stream does not wait for the agent to acknowledge receipt, so a silently dropped connection might result in your agent silently skipping a task. To get a positive signal that the connection is alive, set heartbeatIntervalMs when you open the stream:

heartbeatIntervalMs
Number

The interval in milliseconds at which Lattice sends heartbeat events on the stream. For StreamAsAgent, the default is 30000 ms, and the minimum is 1000 ms. Smaller values are raised to 1000 ms. For ListenAsAgent, heartbeats are disabled unless you set this field.

When set, Lattice sends a heartbeat at the specified interval, and your agent can treat a missing heartbeat as a dropped connection and reconnect. Heartbeats arrive on the same stream as task requests and carry only a timestamp, so the receive loop skips them rather than treating them as tasks.

Replace AGENT_ID with the ID of the agent you want to task. If you are developing on Sandboxes, replace this with the following simulated asset: Demo-Sim-Asset1:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"time"
12
13	Lattice "github.com/anduril/lattice-sdk-go/v5"
14	"github.com/anduril/lattice-sdk-go/v5/client"
15	"github.com/anduril/lattice-sdk-go/v5/option"
16)
17
18// Type URL of the Objective task defined in objective.proto.
19const objectiveType = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"
20
21// parseObjective parses the Objective carried in the task's specification. Over
22// REST, the specification is a google.protobuf.Any represented as JSON: a Type
23// URL alongside the task's flattened fields, which the SDK surfaces as
24// ExtraProperties.
25func parseObjective(specification *Lattice.GoogleProtobufAny) {
26	if specification == nil || specification.Type == nil || *specification.Type != objectiveType {
27		log.Printf("Unsupported task type: %v", specification.GetType())
28		return
29	}
30
31	// Objective is a oneof: read whichever target the operator set.
32	fields := specification.ExtraProperties
33	if entityID, ok := fields["entityId"].(string); ok && entityID != "" {
34		log.Printf("  Objective: entity %s", entityID)
35	} else if lla, ok := fields["lla"].(map[string]interface{}); ok {
36		log.Printf("  Objective: LLA (lat %v, lon %v, alt %vm)",
37			lla["latitudeDegrees"], lla["longitudeDegrees"], lla["altitudeHaeM"])
38	} else {
39		log.Printf("  Objective: no target set")
40	}
41}
42
43func main() {
44	// Get environment variables
45	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
46	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
47	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
48
49	// Check required environment variables
50	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
51		fmt.Println("Missing required environment variables")
52		os.Exit(1)
53	}
54
55	// Initialize headers for sandbox authorization
56	headers := http.Header{}
57	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
58	// Create the client
59	LatticeClient := client.NewClient(
60		option.WithToken(environmentToken),
61		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
62		option.WithHTTPHeader(headers),
63	)
64
65	// Set the entity ID to listen for tasks
66	entityId := "<AGENT_ID>"
67	fmt.Printf("Streaming tasks for entity %s...\n", entityId)
68
69	// Create context for the request
70	ctx := context.Background()
71
72	// Create agent stream request. Set HeartbeatIntervalMs so Lattice sends
73	// periodic heartbeats; a missing heartbeat signals a dropped connection.
74	heartbeatIntervalMs := 30000
75	agentStreamRequest := Lattice.AgentStreamRequest{
76		AgentSelector: &Lattice.EntityIDsSelector{
77			EntityIDs: []string{entityId},
78		},
79		HeartbeatIntervalMs: &heartbeatIntervalMs,
80	}
81
82	// Stream tasks
83	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
84	if err != nil {
85		fmt.Printf("Error streaming tasks: %v\n", err)
86		os.Exit(1)
87	}
88
89	// Process stream events
90	for {
91		select {
92		case <-ctx.Done():
93			log.Printf("Context canceled: %v", ctx.Err())
94			return
95		default:
96			// Continue processing
97		}
98
99		event, err := stream.Recv()
100
101		if errors.Is(err, io.EOF) {
102			log.Println("Stream completed successfully.")
103			return
104		}
105
106		if err != nil {
107			log.Printf("Error receiving message: %v", err)
108			continue
109		}
110
111		if event.Event == "heartbeat" {
112			timestamp := *event.Heartbeat.Timestamp
113			log.Printf("Heartbeat: %s", timestamp)
114		} else {
115			request := event.GetAgentRequest()
116			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
117				task := executeRequest.GetTask()
118				if task != nil {
119					taskId := *task.GetVersion().GetTaskID()
120					taskStatusVersion := *task.GetVersion().GetStatusVersion()
121					description := *task.GetDescription()
122
123					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
124
125					// Parse the Objective the operator sent with the task.
126					parseObjective(task.GetSpecification())
127
128					// Update task status to STATUS_EXECUTING
129					result, err := startTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
130					if err != nil {
131						log.Printf("Error starting task: %v", err)
132						continue
133					}
134
135					log.Printf("Started task with status version: %d", *result.StatusVersion)
136				}
137			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
138				taskToComplete := completeRequest.GetTaskID()
139				if taskToComplete != nil {
140					log.Printf("Completing task: %s", *taskToComplete)
141				}
142			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
143				taskToCancel := cancelRequest.GetTaskID()
144				if taskToCancel != nil {
145					log.Printf("Cancelling task: %s", *taskToCancel)
146				}
147			}
148		}
149
150		// Sleep briefly to prevent tight looping
151		time.Sleep(100 * time.Millisecond)
152	}
153}
154
155// startTask updates the task status to STATUS_EXECUTING
156func startTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
157	// Increment status version for the update
158	taskStatusVersion++
159
160	// Create system principal with the agent entity ID
161	principal := Lattice.Principal{
162		System: &Lattice.System{
163			EntityID: &agentEntityId,
164		},
165	}
166
167	taskStatus := Lattice.TaskStatus{
168		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
169	}
170
171	// Create task status update request
172	taskStatusUpdate := Lattice.TaskStatusUpdate{
173		TaskID:        taskId,
174		StatusVersion: &taskStatusVersion,
175		NewStatus:     &taskStatus,
176		Author:        &principal,
177	}
178
179	// Call the UpdateTaskStatus API
180	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
181	if err != nil {
182		return nil, fmt.Errorf("error updating task status: %w", err)
183	}
184
185	return task.Version, nil
186}

If successful, you see the following output:

$Listening for tasks for entity with ID <entity-id>...

An agent should also handle stream interruptions gracefully. If the connection to Lattice fails, the stream closes. Handle stream errors and reconnect so that your agent keeps its subscription to task updates through transient network issues.

For more information, see Retry connections.

This example reports STATUS_EXECUTING as soon as a task arrives, but an agent isn’t obligated to accept every task.

Before executing the task, validate that the agent can actually perform the task — for example, that the requested parameters are within its capabilities. If it can’t, the agent can reject the task by reporting STATUS_DONE_NOT_OK with a TaskError of ERROR_CODE_REJECTED instead of starting execution.

2

Parse the task specification

Decode the task’s specification to read the Objective the operator sent. Over REST, the specification is delivered as JSON — a type URL alongside the task’s flattened fields. Over gRPC, it’s a protobuf Any that you unpack into the Objective message generated from your message. Check the type URL before reading the fields, since an agent’s catalog can advertise more than one task type. Because Objective is a oneof, inspect which target is set — an entity_id or an lla point:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"time"
12
13	Lattice "github.com/anduril/lattice-sdk-go/v5"
14	"github.com/anduril/lattice-sdk-go/v5/client"
15	"github.com/anduril/lattice-sdk-go/v5/option"
16)
17
18// Type URL of the Objective task defined in objective.proto.
19const objectiveType = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"
20
21// parseObjective parses the Objective carried in the task's specification. Over
22// REST, the specification is a google.protobuf.Any represented as JSON: a Type
23// URL alongside the task's flattened fields, which the SDK surfaces as
24// ExtraProperties.
25func parseObjective(specification *Lattice.GoogleProtobufAny) {
26	if specification == nil || specification.Type == nil || *specification.Type != objectiveType {
27		log.Printf("Unsupported task type: %v", specification.GetType())
28		return
29	}
30
31	// Objective is a oneof: read whichever target the operator set.
32	fields := specification.ExtraProperties
33	if entityID, ok := fields["entityId"].(string); ok && entityID != "" {
34		log.Printf("  Objective: entity %s", entityID)
35	} else if lla, ok := fields["lla"].(map[string]interface{}); ok {
36		log.Printf("  Objective: LLA (lat %v, lon %v, alt %vm)",
37			lla["latitudeDegrees"], lla["longitudeDegrees"], lla["altitudeHaeM"])
38	} else {
39		log.Printf("  Objective: no target set")
40	}
41}
42
43func main() {
44	// Get environment variables
45	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
46	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
47	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
48
49	// Check required environment variables
50	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
51		fmt.Println("Missing required environment variables")
52		os.Exit(1)
53	}
54
55	// Initialize headers for sandbox authorization
56	headers := http.Header{}
57	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
58	// Create the client
59	LatticeClient := client.NewClient(
60		option.WithToken(environmentToken),
61		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
62		option.WithHTTPHeader(headers),
63	)
64
65	// Set the entity ID to listen for tasks
66	entityId := "<AGENT_ID>"
67	fmt.Printf("Streaming tasks for entity %s...\n", entityId)
68
69	// Create context for the request
70	ctx := context.Background()
71
72	// Create agent stream request. Set HeartbeatIntervalMs so Lattice sends
73	// periodic heartbeats; a missing heartbeat signals a dropped connection.
74	heartbeatIntervalMs := 30000
75	agentStreamRequest := Lattice.AgentStreamRequest{
76		AgentSelector: &Lattice.EntityIDsSelector{
77			EntityIDs: []string{entityId},
78		},
79		HeartbeatIntervalMs: &heartbeatIntervalMs,
80	}
81
82	// Stream tasks
83	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
84	if err != nil {
85		fmt.Printf("Error streaming tasks: %v\n", err)
86		os.Exit(1)
87	}
88
89	// Process stream events
90	for {
91		select {
92		case <-ctx.Done():
93			log.Printf("Context canceled: %v", ctx.Err())
94			return
95		default:
96			// Continue processing
97		}
98
99		event, err := stream.Recv()
100
101		if errors.Is(err, io.EOF) {
102			log.Println("Stream completed successfully.")
103			return
104		}
105
106		if err != nil {
107			log.Printf("Error receiving message: %v", err)
108			continue
109		}
110
111		if event.Event == "heartbeat" {
112			timestamp := *event.Heartbeat.Timestamp
113			log.Printf("Heartbeat: %s", timestamp)
114		} else {
115			request := event.GetAgentRequest()
116			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
117				task := executeRequest.GetTask()
118				if task != nil {
119					taskId := *task.GetVersion().GetTaskID()
120					taskStatusVersion := *task.GetVersion().GetStatusVersion()
121					description := *task.GetDescription()
122
123					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
124
125					// Parse the Objective the operator sent with the task.
126					parseObjective(task.GetSpecification())
127
128					// Update task status to STATUS_EXECUTING
129					result, err := startTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
130					if err != nil {
131						log.Printf("Error starting task: %v", err)
132						continue
133					}
134
135					log.Printf("Started task with status version: %d", *result.StatusVersion)
136				}
137			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
138				taskToComplete := completeRequest.GetTaskID()
139				if taskToComplete != nil {
140					log.Printf("Completing task: %s", *taskToComplete)
141				}
142			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
143				taskToCancel := cancelRequest.GetTaskID()
144				if taskToCancel != nil {
145					log.Printf("Cancelling task: %s", *taskToCancel)
146				}
147			}
148		}
149
150		// Sleep briefly to prevent tight looping
151		time.Sleep(100 * time.Millisecond)
152	}
153}
154
155// startTask updates the task status to STATUS_EXECUTING
156func startTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
157	// Increment status version for the update
158	taskStatusVersion++
159
160	// Create system principal with the agent entity ID
161	principal := Lattice.Principal{
162		System: &Lattice.System{
163			EntityID: &agentEntityId,
164		},
165	}
166
167	taskStatus := Lattice.TaskStatus{
168		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
169	}
170
171	// Create task status update request
172	taskStatusUpdate := Lattice.TaskStatusUpdate{
173		TaskID:        taskId,
174		StatusVersion: &taskStatusVersion,
175		NewStatus:     &taskStatus,
176		Author:        &principal,
177	}
178
179	// Call the UpdateTaskStatus API
180	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
181	if err != nil {
182		return nil, fmt.Errorf("error updating task status: %w", err)
183	}
184
185	return task.Version, nil
186}

The gRPC examples import an Objective type generated from your own objective.proto. Publish your schema to the Schema Registry, then generate the language bindings before running the agent.

Task and verify the task handler

The following steps aren’t required to implement tasking — they show how to exercise your task handler end to end in a development environment. Assign a task to your agent, then confirm it receives, parses, and updates the task.

1

Assign a task using the UI

In most cases an operator uses the Lattice UI to task your agent. To test the agent’s task handler in Lattice Sandboxes, do the following:

  1. Open your environment’s Lattice UI, and choose an asset from the Assets panel. On Sandboxes, choose Demo-Sim-Asset1.
  2. From the entity pane, choose Task, then select the Objective task.
  3. From the Task Details panel on the right hand side, set the objective — either an entity to reconnoiter or a lat/lon/altitude point — then choose Execute Task. This is the same field your agent reads from the task specification.
2

Verify the task status

If successful, you see the parsed task data and the updated status version in your local development console:

$Starting task <task-id>, version 1: Reconnaissance objective
$  Objective: LLA (lat 37.7749, lon -122.4194, alt 100.0m)
$Started task with status version: 2.

The agent has successfully parsed the task data, updated the task status, and incremented the task status version.

Update the status of a task

As an agent makes progress, it reports real-time updates to Lattice with UpdateTaskStatus, incrementing the task’s statusVersion on each update. In Lattice, tasks move through the following states:

1

STATUS_SENT

Lattice automatically starts with STATUS_SENT when it sends a task to an agent:

1"task": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 1
6    },
7    "status": {
8        "status": "STATUS_SENT"
9    }
10}
2

STATUS_MACHINE_RECEIPT

The agent then responds back with status STATUS_MACHINE_RECEIPT, indicating that the task has been received, and incrementing statusVersion accordingly:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 2
6    },
7    "status": {
8        "status": "STATUS_MACHINE_RECEIPT"
9    }
10}
3

STATUS_ACK

When the agent is ready to acknowledge the task, it does so using STATUS_ACK, and again increments statusVersion:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 3
6    },
7    "status": {
8        "status": "STATUS_ACK"
9    }
10}
4

STATUS_WILCO

The agent confirms it intends to execute the task using STATUS_WILCO:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 4
6    },
7    "status": {
8        "status": "STATUS_WILCO"
9    }
10}
5

STATUS_EXECUTING

As the agent begins to actively execute the task, it indicates this by reporting STATUS_EXECUTING back to Lattice:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 5
6    },
7    "status": {
8        "status": "STATUS_EXECUTING"
9    }
10}
6

STATUS_DONE_OK

Finally, when the agent reaches a terminal state and completes the task successfully, it reports STATUS_DONE_OK. The agent can reach this state on its own, such as when its logic determines that the task’s objective is met, or in response to an operator-initiated request for task completion:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 6
6    },
7    "status": {
8        "status": "STATUS_DONE_OK"
9    }
10}
7

STATUS_DONE_NOT_OK

If the agent reaches a terminal state but does not complete the task successfully, it reports STATUS_DONE_NOT_OK. The agent can reach this state on its own, such as when its logic determines that the task can’t be completed, or in response to an operator-initiated request for task completion or cancellation. The agent should include a descriptive TaskError when reporting STATUS_DONE_NOT_OK:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 7
6    },
7    "status": {
8        "status": "STATUS_DONE_NOT_OK",
9        "taskError": {
10        "code": "ERROR_CODE_FAILED",
11        "message": "The asset failed task execution due to an internal error.",
12        }
13    }
14}

In this example, the message indicates that the agent encountered an internal error during task execution. You can add more descriptive errors to help the operator troubleshoot the issue accordingly.

The STATUS_DONE_OK and STATUS_DONE_NOT_OK statuses are considered terminal states. Once a task reaches either state, it’s complete and cannot be updated.

Handle task cancellation

An operator can request that a task be cancelled while the agent is executing it. When a task has already been sent to an agent, Lattice routes the CancelTask request to that agent, which decides whether to accept or reject it. For the operator’s side of this workflow, see Cancel tasks.

1

Set up a task processor

Create a task processor that listens for tasks assigned to your agent and handles cancellation requests. For demonstration purposes, this example uses a TASK_ACTIVE environment variable to control whether the agent accepts or rejects cancellations:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"strings"
12	"time"
13
14	Lattice "github.com/anduril/lattice-sdk-go/v5"
15	"github.com/anduril/lattice-sdk-go/v5/client"
16	"github.com/anduril/lattice-sdk-go/v5/option"
17)
18
19// Sets whether the task is currently being processed. If so, it cannot be cancelled.
20var taskActive bool
21
22func main() {
23	// Get environment variables
24	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
25	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
26	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
27	taskActiveStr := os.Getenv("TASK_ACTIVE")
28
29	// Check required environment variables
30	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
31		fmt.Println("Missing required environment variables")
32		os.Exit(1)
33	}
34
35	// Parse TASK_ACTIVE environment variable
36	taskActive = strings.ToLower(taskActiveStr) == "true"
37
38	// Initialize headers for sandbox authorization
39	headers := http.Header{}
40	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
41
42	// Create the client
43	LatticeClient := client.NewClient(
44		option.WithToken(environmentToken),
45		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
46		option.WithHTTPHeader(headers),
47	)
48
49	// Set the entity ID to listen for tasks
50	entityId := "<AGENT_ID>"
51	fmt.Printf("Streaming tasks for entity: %s...\n", entityId)
52
53	// Create context for the request
54	ctx := context.Background()
55
56	// Create agent stream request
57	agentStreamRequest := Lattice.AgentStreamRequest{
58		AgentSelector: &Lattice.EntityIDsSelector{
59			EntityIDs: []string{entityId},
60		},
61	}
62
63	// Stream tasks
64	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
65	if err != nil {
66		fmt.Printf("Error streaming tasks: %v\n", err)
67		os.Exit(1)
68	}
69
70	// Process stream events
71	for {
72		select {
73		case <-ctx.Done():
74			log.Printf("Context canceled: %v", ctx.Err())
75			return
76		default:
77			// Continue processing
78		}
79
80		event, err := stream.Recv()
81
82		if errors.Is(err, io.EOF) {
83			log.Println("Stream completed successfully.")
84			return
85		}
86
87		if err != nil {
88			log.Printf("Error receiving message: %v", err)
89			continue
90		}
91
92		if event.Event == "heartbeat" {
93			timestamp := *event.Heartbeat.Timestamp
94			log.Printf("Heartbeat: %s", timestamp)
95		} else {
96			request := event.GetAgentRequest()
97			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
98				task := executeRequest.GetTask()
99				if task != nil {
100					taskId := *task.GetVersion().GetTaskID()
101					taskStatusVersion := *task.GetVersion().GetStatusVersion()
102					description := *task.GetDescription()
103
104					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
105
106					// Update task status to STATUS_EXECUTING
107					result, err := executeTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
108					if err != nil {
109						log.Printf("Error starting task: %v", err)
110						continue
111					}
112
113					log.Printf("Started task with status version: %d", *result.StatusVersion)
114				}
115			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
116				taskToComplete := completeRequest.GetTaskID()
117				if taskToComplete != nil {
118					log.Printf("Completing task: %s", *taskToComplete)
119					err := completeTask(ctx, LatticeClient, *taskToComplete, entityId)
120					if err != nil {
121						log.Printf("Error completing task: %v", err)
122					}
123				}
124			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
125				taskToCancel := cancelRequest.GetTaskID()
126				if taskToCancel != nil {
127					log.Printf("Cancelling task: %s", *taskToCancel)
128					err := cancelTask(ctx, LatticeClient, *taskToCancel, entityId)
129					if err != nil {
130						log.Printf("Error cancelling task: %v", err)
131					}
132				}
133			}
134		}
135
136		// Sleep briefly to prevent tight looping
137		time.Sleep(100 * time.Millisecond)
138	}
139}
140
141// executeTask updates the task status to STATUS_EXECUTING
142func executeTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
143	// Increment status version for the update
144	taskStatusVersion++
145
146	// Create system principal with the agent entity ID
147	principal := Lattice.Principal{
148		System: &Lattice.System{
149			EntityID: &agentEntityId,
150		},
151	}
152
153	taskStatus := Lattice.TaskStatus{
154		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
155	}
156
157	// Create task status update request
158	taskStatusUpdate := Lattice.TaskStatusUpdate{
159		TaskID:        taskId,
160		StatusVersion: &taskStatusVersion,
161		NewStatus:     &taskStatus,
162		Author:        &principal,
163	}
164
165	// Call the UpdateTaskStatus API
166	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
167	if err != nil {
168		return nil, fmt.Errorf("error updating task status: %w", err)
169	}
170
171	return task.Version, nil
172}
173
174// cancelTask handles cancellation requests from Lattice
175func cancelTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
176	// Get current task to retrieve status_version
177	getTaskRequest := Lattice.GetTaskRequest{
178		TaskID: taskId,
179	}
180	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
181	if err != nil {
182		return fmt.Errorf("error getting task: %w", err)
183	}
184	if task.Status == nil || task.Version == nil || task.Version.StatusVersion == nil {
185		return fmt.Errorf("task status or version is missing")
186	}
187	currentTaskStatus := task.Status.Status
188	taskStatusVersion := *task.Version.StatusVersion
189	taskStatusVersion++
190
191	// Create system principal with the agent entity ID
192	principal := Lattice.Principal{
193		System: &Lattice.System{
194			EntityID: &entityId,
195		},
196	}
197
198	if taskActive {
199		// Reject cancellation: task is active and cannot be cancelled
200		rejectedMessage := "Task is already active, and cannot be cancelled."
201		taskStatus := Lattice.TaskStatus{
202			// Because the cancellation is being rejected, we do not
203			// change the task status.
204			Status: currentTaskStatus,
205			TaskError: &Lattice.TaskError{
206				Code:    Lattice.TaskErrorCodeErrorCodeRejected.Ptr(),
207				Message: &rejectedMessage,
208			},
209		}
210
211		taskStatusUpdate := Lattice.TaskStatusUpdate{
212			TaskID:        taskId,
213			StatusVersion: &taskStatusVersion,
214			NewStatus:     &taskStatus,
215			Author:        &principal,
216		}
217
218		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
219		if err != nil {
220			return fmt.Errorf("error updating task status: %w", err)
221		}
222
223		log.Println("Task could not be cancelled.")
224	} else {
225		// Accept cancellation
226		cancelledMessage := "Task cancelled by agent."
227		taskStatus := Lattice.TaskStatus{
228			Status: Lattice.TaskStatusStatusStatusDoneNotOk.Ptr(),
229			TaskError: &Lattice.TaskError{
230				Code:    Lattice.TaskErrorCodeErrorCodeCancelled.Ptr(),
231				Message: &cancelledMessage,
232			},
233		}
234
235		taskStatusUpdate := Lattice.TaskStatusUpdate{
236			TaskID:        taskId,
237			StatusVersion: &taskStatusVersion,
238			NewStatus:     &taskStatus,
239			Author:        &principal,
240		}
241
242		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
243		if err != nil {
244			return fmt.Errorf("error updating task status: %w", err)
245		}
246
247		log.Println("Task has been cancelled.")
248	}
249
250	return nil
251}
252
253// completeTask updates the task status to STATUS_DONE_OK
254func completeTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
255	// Get current task to retrieve status_version
256	getTaskRequest := Lattice.GetTaskRequest{
257		TaskID: taskId,
258	}
259	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
260	if err != nil {
261		return fmt.Errorf("error getting task: %w", err)
262	}
263
264	if task.Version == nil || task.Version.StatusVersion == nil {
265		return fmt.Errorf("task version is missing")
266	}
267
268	taskStatusVersion := *task.Version.StatusVersion
269	// Increment version and update to terminal state
270	taskStatusVersion++
271
272	// Create system principal with the agent entity ID
273	principal := Lattice.Principal{
274		System: &Lattice.System{
275			EntityID: &entityId,
276		},
277	}
278
279	taskStatus := Lattice.TaskStatus{
280		Status: Lattice.TaskStatusStatusStatusDoneOk.Ptr(),
281	}
282
283	taskStatusUpdate := Lattice.TaskStatusUpdate{
284		TaskID:        taskId,
285		StatusVersion: &taskStatusVersion,
286		NewStatus:     &taskStatus,
287		Author:        &principal,
288	}
289
290	_, err = client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
291	if err != nil {
292		return fmt.Errorf("error updating task status: %w", err)
293	}
294
295	return nil
296}

The StreamAsAgent API establishes a server-sent events (SSE) stream that delivers three types of requests: executeRequest which notifies the agent to start executing a task, completeRequest which requests the agent to complete a task, and cancelRequest which requests the agent to cancel a task.

2

Handle cancellation requests

When a cancelRequest arrives, the agent retrieves the current task state and decides whether to accept or reject the cancellation.

Rejecting cancellation:

If the task is active and cannot be cancelled, the agent rejects the cancellation by keeping the current status and attaching a TaskError with code ERROR_CODE_REJECTED:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"strings"
12	"time"
13
14	Lattice "github.com/anduril/lattice-sdk-go/v5"
15	"github.com/anduril/lattice-sdk-go/v5/client"
16	"github.com/anduril/lattice-sdk-go/v5/option"
17)
18
19// Sets whether the task is currently being processed. If so, it cannot be cancelled.
20var taskActive bool
21
22func main() {
23	// Get environment variables
24	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
25	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
26	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
27	taskActiveStr := os.Getenv("TASK_ACTIVE")
28
29	// Check required environment variables
30	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
31		fmt.Println("Missing required environment variables")
32		os.Exit(1)
33	}
34
35	// Parse TASK_ACTIVE environment variable
36	taskActive = strings.ToLower(taskActiveStr) == "true"
37
38	// Initialize headers for sandbox authorization
39	headers := http.Header{}
40	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
41
42	// Create the client
43	LatticeClient := client.NewClient(
44		option.WithToken(environmentToken),
45		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
46		option.WithHTTPHeader(headers),
47	)
48
49	// Set the entity ID to listen for tasks
50	entityId := "<AGENT_ID>"
51	fmt.Printf("Streaming tasks for entity: %s...\n", entityId)
52
53	// Create context for the request
54	ctx := context.Background()
55
56	// Create agent stream request
57	agentStreamRequest := Lattice.AgentStreamRequest{
58		AgentSelector: &Lattice.EntityIDsSelector{
59			EntityIDs: []string{entityId},
60		},
61	}
62
63	// Stream tasks
64	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
65	if err != nil {
66		fmt.Printf("Error streaming tasks: %v\n", err)
67		os.Exit(1)
68	}
69
70	// Process stream events
71	for {
72		select {
73		case <-ctx.Done():
74			log.Printf("Context canceled: %v", ctx.Err())
75			return
76		default:
77			// Continue processing
78		}
79
80		event, err := stream.Recv()
81
82		if errors.Is(err, io.EOF) {
83			log.Println("Stream completed successfully.")
84			return
85		}
86
87		if err != nil {
88			log.Printf("Error receiving message: %v", err)
89			continue
90		}
91
92		if event.Event == "heartbeat" {
93			timestamp := *event.Heartbeat.Timestamp
94			log.Printf("Heartbeat: %s", timestamp)
95		} else {
96			request := event.GetAgentRequest()
97			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
98				task := executeRequest.GetTask()
99				if task != nil {
100					taskId := *task.GetVersion().GetTaskID()
101					taskStatusVersion := *task.GetVersion().GetStatusVersion()
102					description := *task.GetDescription()
103
104					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
105
106					// Update task status to STATUS_EXECUTING
107					result, err := executeTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
108					if err != nil {
109						log.Printf("Error starting task: %v", err)
110						continue
111					}
112
113					log.Printf("Started task with status version: %d", *result.StatusVersion)
114				}
115			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
116				taskToComplete := completeRequest.GetTaskID()
117				if taskToComplete != nil {
118					log.Printf("Completing task: %s", *taskToComplete)
119					err := completeTask(ctx, LatticeClient, *taskToComplete, entityId)
120					if err != nil {
121						log.Printf("Error completing task: %v", err)
122					}
123				}
124			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
125				taskToCancel := cancelRequest.GetTaskID()
126				if taskToCancel != nil {
127					log.Printf("Cancelling task: %s", *taskToCancel)
128					err := cancelTask(ctx, LatticeClient, *taskToCancel, entityId)
129					if err != nil {
130						log.Printf("Error cancelling task: %v", err)
131					}
132				}
133			}
134		}
135
136		// Sleep briefly to prevent tight looping
137		time.Sleep(100 * time.Millisecond)
138	}
139}
140
141// executeTask updates the task status to STATUS_EXECUTING
142func executeTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
143	// Increment status version for the update
144	taskStatusVersion++
145
146	// Create system principal with the agent entity ID
147	principal := Lattice.Principal{
148		System: &Lattice.System{
149			EntityID: &agentEntityId,
150		},
151	}
152
153	taskStatus := Lattice.TaskStatus{
154		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
155	}
156
157	// Create task status update request
158	taskStatusUpdate := Lattice.TaskStatusUpdate{
159		TaskID:        taskId,
160		StatusVersion: &taskStatusVersion,
161		NewStatus:     &taskStatus,
162		Author:        &principal,
163	}
164
165	// Call the UpdateTaskStatus API
166	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
167	if err != nil {
168		return nil, fmt.Errorf("error updating task status: %w", err)
169	}
170
171	return task.Version, nil
172}
173
174// cancelTask handles cancellation requests from Lattice
175func cancelTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
176	// Get current task to retrieve status_version
177	getTaskRequest := Lattice.GetTaskRequest{
178		TaskID: taskId,
179	}
180	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
181	if err != nil {
182		return fmt.Errorf("error getting task: %w", err)
183	}
184	if task.Status == nil || task.Version == nil || task.Version.StatusVersion == nil {
185		return fmt.Errorf("task status or version is missing")
186	}
187	currentTaskStatus := task.Status.Status
188	taskStatusVersion := *task.Version.StatusVersion
189	taskStatusVersion++
190
191	// Create system principal with the agent entity ID
192	principal := Lattice.Principal{
193		System: &Lattice.System{
194			EntityID: &entityId,
195		},
196	}
197
198	if taskActive {
199		// Reject cancellation: task is active and cannot be cancelled
200		rejectedMessage := "Task is already active, and cannot be cancelled."
201		taskStatus := Lattice.TaskStatus{
202			// Because the cancellation is being rejected, we do not
203			// change the task status.
204			Status: currentTaskStatus,
205			TaskError: &Lattice.TaskError{
206				Code:    Lattice.TaskErrorCodeErrorCodeRejected.Ptr(),
207				Message: &rejectedMessage,
208			},
209		}
210
211		taskStatusUpdate := Lattice.TaskStatusUpdate{
212			TaskID:        taskId,
213			StatusVersion: &taskStatusVersion,
214			NewStatus:     &taskStatus,
215			Author:        &principal,
216		}
217
218		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
219		if err != nil {
220			return fmt.Errorf("error updating task status: %w", err)
221		}
222
223		log.Println("Task could not be cancelled.")
224	} else {
225		// Accept cancellation
226		cancelledMessage := "Task cancelled by agent."
227		taskStatus := Lattice.TaskStatus{
228			Status: Lattice.TaskStatusStatusStatusDoneNotOk.Ptr(),
229			TaskError: &Lattice.TaskError{
230				Code:    Lattice.TaskErrorCodeErrorCodeCancelled.Ptr(),
231				Message: &cancelledMessage,
232			},
233		}
234
235		taskStatusUpdate := Lattice.TaskStatusUpdate{
236			TaskID:        taskId,
237			StatusVersion: &taskStatusVersion,
238			NewStatus:     &taskStatus,
239			Author:        &principal,
240		}
241
242		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
243		if err != nil {
244			return fmt.Errorf("error updating task status: %w", err)
245		}
246
247		log.Println("Task has been cancelled.")
248	}
249
250	return nil
251}
252
253// completeTask updates the task status to STATUS_DONE_OK
254func completeTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
255	// Get current task to retrieve status_version
256	getTaskRequest := Lattice.GetTaskRequest{
257		TaskID: taskId,
258	}
259	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
260	if err != nil {
261		return fmt.Errorf("error getting task: %w", err)
262	}
263
264	if task.Version == nil || task.Version.StatusVersion == nil {
265		return fmt.Errorf("task version is missing")
266	}
267
268	taskStatusVersion := *task.Version.StatusVersion
269	// Increment version and update to terminal state
270	taskStatusVersion++
271
272	// Create system principal with the agent entity ID
273	principal := Lattice.Principal{
274		System: &Lattice.System{
275			EntityID: &entityId,
276		},
277	}
278
279	taskStatus := Lattice.TaskStatus{
280		Status: Lattice.TaskStatusStatusStatusDoneOk.Ptr(),
281	}
282
283	taskStatusUpdate := Lattice.TaskStatusUpdate{
284		TaskID:        taskId,
285		StatusVersion: &taskStatusVersion,
286		NewStatus:     &taskStatus,
287		Author:        &principal,
288	}
289
290	_, err = client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
291	if err != nil {
292		return fmt.Errorf("error updating task status: %w", err)
293	}
294
295	return nil
296}

Accepting cancellation:

If the task can be cancelled, the agent accepts by setting the status to STATUS_DONE_NOT_OK with a TaskError indicating ERROR_CODE_CANCELLED:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"strings"
12	"time"
13
14	Lattice "github.com/anduril/lattice-sdk-go/v5"
15	"github.com/anduril/lattice-sdk-go/v5/client"
16	"github.com/anduril/lattice-sdk-go/v5/option"
17)
18
19// Sets whether the task is currently being processed. If so, it cannot be cancelled.
20var taskActive bool
21
22func main() {
23	// Get environment variables
24	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
25	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
26	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
27	taskActiveStr := os.Getenv("TASK_ACTIVE")
28
29	// Check required environment variables
30	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
31		fmt.Println("Missing required environment variables")
32		os.Exit(1)
33	}
34
35	// Parse TASK_ACTIVE environment variable
36	taskActive = strings.ToLower(taskActiveStr) == "true"
37
38	// Initialize headers for sandbox authorization
39	headers := http.Header{}
40	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
41
42	// Create the client
43	LatticeClient := client.NewClient(
44		option.WithToken(environmentToken),
45		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
46		option.WithHTTPHeader(headers),
47	)
48
49	// Set the entity ID to listen for tasks
50	entityId := "<AGENT_ID>"
51	fmt.Printf("Streaming tasks for entity: %s...\n", entityId)
52
53	// Create context for the request
54	ctx := context.Background()
55
56	// Create agent stream request
57	agentStreamRequest := Lattice.AgentStreamRequest{
58		AgentSelector: &Lattice.EntityIDsSelector{
59			EntityIDs: []string{entityId},
60		},
61	}
62
63	// Stream tasks
64	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
65	if err != nil {
66		fmt.Printf("Error streaming tasks: %v\n", err)
67		os.Exit(1)
68	}
69
70	// Process stream events
71	for {
72		select {
73		case <-ctx.Done():
74			log.Printf("Context canceled: %v", ctx.Err())
75			return
76		default:
77			// Continue processing
78		}
79
80		event, err := stream.Recv()
81
82		if errors.Is(err, io.EOF) {
83			log.Println("Stream completed successfully.")
84			return
85		}
86
87		if err != nil {
88			log.Printf("Error receiving message: %v", err)
89			continue
90		}
91
92		if event.Event == "heartbeat" {
93			timestamp := *event.Heartbeat.Timestamp
94			log.Printf("Heartbeat: %s", timestamp)
95		} else {
96			request := event.GetAgentRequest()
97			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
98				task := executeRequest.GetTask()
99				if task != nil {
100					taskId := *task.GetVersion().GetTaskID()
101					taskStatusVersion := *task.GetVersion().GetStatusVersion()
102					description := *task.GetDescription()
103
104					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
105
106					// Update task status to STATUS_EXECUTING
107					result, err := executeTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
108					if err != nil {
109						log.Printf("Error starting task: %v", err)
110						continue
111					}
112
113					log.Printf("Started task with status version: %d", *result.StatusVersion)
114				}
115			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
116				taskToComplete := completeRequest.GetTaskID()
117				if taskToComplete != nil {
118					log.Printf("Completing task: %s", *taskToComplete)
119					err := completeTask(ctx, LatticeClient, *taskToComplete, entityId)
120					if err != nil {
121						log.Printf("Error completing task: %v", err)
122					}
123				}
124			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
125				taskToCancel := cancelRequest.GetTaskID()
126				if taskToCancel != nil {
127					log.Printf("Cancelling task: %s", *taskToCancel)
128					err := cancelTask(ctx, LatticeClient, *taskToCancel, entityId)
129					if err != nil {
130						log.Printf("Error cancelling task: %v", err)
131					}
132				}
133			}
134		}
135
136		// Sleep briefly to prevent tight looping
137		time.Sleep(100 * time.Millisecond)
138	}
139}
140
141// executeTask updates the task status to STATUS_EXECUTING
142func executeTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
143	// Increment status version for the update
144	taskStatusVersion++
145
146	// Create system principal with the agent entity ID
147	principal := Lattice.Principal{
148		System: &Lattice.System{
149			EntityID: &agentEntityId,
150		},
151	}
152
153	taskStatus := Lattice.TaskStatus{
154		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
155	}
156
157	// Create task status update request
158	taskStatusUpdate := Lattice.TaskStatusUpdate{
159		TaskID:        taskId,
160		StatusVersion: &taskStatusVersion,
161		NewStatus:     &taskStatus,
162		Author:        &principal,
163	}
164
165	// Call the UpdateTaskStatus API
166	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
167	if err != nil {
168		return nil, fmt.Errorf("error updating task status: %w", err)
169	}
170
171	return task.Version, nil
172}
173
174// cancelTask handles cancellation requests from Lattice
175func cancelTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
176	// Get current task to retrieve status_version
177	getTaskRequest := Lattice.GetTaskRequest{
178		TaskID: taskId,
179	}
180	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
181	if err != nil {
182		return fmt.Errorf("error getting task: %w", err)
183	}
184	if task.Status == nil || task.Version == nil || task.Version.StatusVersion == nil {
185		return fmt.Errorf("task status or version is missing")
186	}
187	currentTaskStatus := task.Status.Status
188	taskStatusVersion := *task.Version.StatusVersion
189	taskStatusVersion++
190
191	// Create system principal with the agent entity ID
192	principal := Lattice.Principal{
193		System: &Lattice.System{
194			EntityID: &entityId,
195		},
196	}
197
198	if taskActive {
199		// Reject cancellation: task is active and cannot be cancelled
200		rejectedMessage := "Task is already active, and cannot be cancelled."
201		taskStatus := Lattice.TaskStatus{
202			// Because the cancellation is being rejected, we do not
203			// change the task status.
204			Status: currentTaskStatus,
205			TaskError: &Lattice.TaskError{
206				Code:    Lattice.TaskErrorCodeErrorCodeRejected.Ptr(),
207				Message: &rejectedMessage,
208			},
209		}
210
211		taskStatusUpdate := Lattice.TaskStatusUpdate{
212			TaskID:        taskId,
213			StatusVersion: &taskStatusVersion,
214			NewStatus:     &taskStatus,
215			Author:        &principal,
216		}
217
218		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
219		if err != nil {
220			return fmt.Errorf("error updating task status: %w", err)
221		}
222
223		log.Println("Task could not be cancelled.")
224	} else {
225		// Accept cancellation
226		cancelledMessage := "Task cancelled by agent."
227		taskStatus := Lattice.TaskStatus{
228			Status: Lattice.TaskStatusStatusStatusDoneNotOk.Ptr(),
229			TaskError: &Lattice.TaskError{
230				Code:    Lattice.TaskErrorCodeErrorCodeCancelled.Ptr(),
231				Message: &cancelledMessage,
232			},
233		}
234
235		taskStatusUpdate := Lattice.TaskStatusUpdate{
236			TaskID:        taskId,
237			StatusVersion: &taskStatusVersion,
238			NewStatus:     &taskStatus,
239			Author:        &principal,
240		}
241
242		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
243		if err != nil {
244			return fmt.Errorf("error updating task status: %w", err)
245		}
246
247		log.Println("Task has been cancelled.")
248	}
249
250	return nil
251}
252
253// completeTask updates the task status to STATUS_DONE_OK
254func completeTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
255	// Get current task to retrieve status_version
256	getTaskRequest := Lattice.GetTaskRequest{
257		TaskID: taskId,
258	}
259	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
260	if err != nil {
261		return fmt.Errorf("error getting task: %w", err)
262	}
263
264	if task.Version == nil || task.Version.StatusVersion == nil {
265		return fmt.Errorf("task version is missing")
266	}
267
268	taskStatusVersion := *task.Version.StatusVersion
269	// Increment version and update to terminal state
270	taskStatusVersion++
271
272	// Create system principal with the agent entity ID
273	principal := Lattice.Principal{
274		System: &Lattice.System{
275			EntityID: &entityId,
276		},
277	}
278
279	taskStatus := Lattice.TaskStatus{
280		Status: Lattice.TaskStatusStatusStatusDoneOk.Ptr(),
281	}
282
283	taskStatusUpdate := Lattice.TaskStatusUpdate{
284		TaskID:        taskId,
285		StatusVersion: &taskStatusVersion,
286		NewStatus:     &taskStatus,
287		Author:        &principal,
288	}
289
290	_, err = client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
291	if err != nil {
292		return fmt.Errorf("error updating task status: %w", err)
293	}
294
295	return nil
296}

The agent must first retrieve the current task using GetTask to obtain the current statusVersion, then increment it before calling UpdateTaskStatus.

What’s next